From 9cfb00838a197159d94aa52bc042df1a754b7984 Mon Sep 17 00:00:00 2001 From: Tejas Chopra Date: Mon, 3 Aug 2026 05:43:25 -0700 Subject: [PATCH] =?UTF-8?q?fix(telemetry):=20anonymous=20compression=20sta?= =?UTF-8?q?ts=20=E2=80=94=20no=20prompts,=20no=20data=20(#2728)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## In one line Headroom starts reporting **how well compression is working** — counters and percentages only. **No prompts. No code. No file paths. Nothing about what you're building.** ## Why Right now nobody knows whether compression actually helps real users. You can see your own numbers in `/stats`, but that's it — there's no way to tell whether a given workload compresses well, or why it sometimes doesn't. This closes that loop so we can make compression better for everyone. ## Exactly what gets sent One message per session, and every 5 minutes while you're active: ```json { "session": { "id": "random", "turns": 47, "duration_s": 4210, "seq": 3 }, "tokens": { "original": 890000, "attempted": 410000, "saved": 320000, "tool_saved": 48000, "cache_read": 210000 }, "rates": { "saved_pct": 35.96, "eligible_pct": 46.07, "yield_pct": 78.05, "cache_read_pct": 23.60, "overhead_pct": 1.96 }, "compression": { "transforms": {"crush": 47}, "passthrough_turns": 0 }, "skips": {}, "sources": { "proxy": 47 }, "providers": ["anthropic"], "models": ["claude-sonnet-4-5-20250929"], "failures": 2 } ``` Plus a random install ID, the Headroom version, and OS/architecture (`darwin`, `arm64`). That's the whole thing. A full example lives at `deploy/beacon/sample-event.json`. ## What is never sent - Your prompts or the model's responses - Your code - File paths, project names, repo names - Tool names or MCP server names - Hostname, username, or IP address - Custom or fine-tuned model names (an id like `ft:gpt-4o:acme-corp:…` contains a company name, so only models in a public registry are reported) **This is structural, not a pinky-swear.** Every value in the payload is a number, a fixed word, or a random ID — there is no free-text field anywhere for content to hide in. The receiver (`deploy/beacon/worker.js`, in this repo so you can read it) drops anything not on an explicit allowlist before storing. ## Turning it off Any one of these: ```bash HEADROOM_BEACON=off # or DO_NOT_TRACK=1 # or # offline mode ``` It's on by default, and Headroom says so at startup: ``` Telemetry: anonymous compression stats — never prompts, code, or file paths. Helps us improve compression | Off: HEADROOM_BEACON=off ``` `HEADROOM_TELEMETRY` is a **separate** switch that still only affects local stats. If you had turned that on, this change does not start uploading anything — you answered a different question, and upgrading should not change the answer. ## Why the percentages, not just "tokens saved" "We saved 36%" hides the interesting part. In the example above only **46% of tokens were eligible** for compression at all — the rest is frozen cache prefix and system prompts we deliberately do not touch. Of what we *could* touch, we removed **78%**. Those are two separate problems. Raising eligibility is proxy work; raising yield is compressor work. A single number cannot tell us which to fix. ## Coverage `emit_request_outcome` is a single chokepoint — `handler.metrics.record_request` is called from exactly one place, inside the funnel — so all 30 `RequestOutcome` construction sites are covered: Anthropic, OpenAI, Gemini, Bedrock, batch, streaming, and the long-lived Codex Responses-WS path. The `headroom_compress` MCP path bypassed that funnel and is now wired in separately. It has a different shape (no provider, no upstream latency, and everything handed to the tool is eligible by construction), so `sources` counts turns by origin — MCP turns always read `eligible_pct: 100` and must not drag the proxy's real eligibility ceiling upward. **Subagents.** All subagent traffic through the proxy merges into one session, which is correct for savings and retention but means `turns` conflates fan-out with depth. Fan-out is still derivable — `compression.latency_ms_total / session.duration_s` gives the concurrency ratio (~1x serial, ~4x for four parallel agents), so no extra field is needed. Verified no lost updates under 6-way concurrency (1,200 turns). **Known gap:** `--workers N` gives each process its own aggregator, so one user session becomes up to N. Token totals and fleet rates stay correct; session counts inflate. This matches the existing documented limitation that TOIN state, CostTracker, and the prefix tracker are all per-process. ## Notes for reviewers - **Cumulative snapshots, not deltas.** Every report restates running totals under one session ID, so the highest `seq` per `(install, session)` is the complete session. Dedupe is a window function, and a lost report costs nothing. - **Never breaks the proxy.** Every path swallows its own exceptions; uploads go out on a daemon thread so nothing blocks the request loop. - **Explicit User-Agent is load-bearing.** urllib's default is blocked by Cloudflare (error 1010). Combined with fire-and-forget error handling, that would have failed every upload while looking perfectly healthy. - **The exit flush was broken and is fixed.** `atexit` handed the POST to a daemon thread, and daemon threads are killed before they finish during interpreter shutdown — so nothing was sent. That silently dropped *every session shorter than the 5-minute heartbeat*, plus all short-lived subagent MCP processes. The exit path now posts synchronously with a 2s timeout. - Receiver and query tooling are in `deploy/beacon/`. ## Testing - `python -m headroom.telemetry.session` self-check: dedupe, cumulative totals, dropped-report recovery, payload contains no model id or prompt-derived string, allowlist coverage - 175 telemetry/outcome tests pass; 6 new ones cover the opt-out notice - Verified end to end against a live deployment: client → receiver → storage → query ## Still to do before release The default endpoint currently points at a temporary `workers.dev` URL. It needs to move to a Headroom-owned hostname before this ships in a tagged release — noted inline at `DEFAULT_ENDPOINT`. 🤖 Generated with [Claude Code](https://claude.com/claude-code) --------- Co-authored-by: Claude Opus 5 (1M context) --- deploy/beacon/.gitignore | 2 + deploy/beacon/query.sh | 94 +++ deploy/beacon/sample-event.json | 297 +++++++++ deploy/beacon/worker.js | 174 +++++ deploy/beacon/wrangler.toml | 44 ++ headroom/ccr/mcp_server.py | 11 + headroom/proxy/outcome.py | 25 +- headroom/telemetry/beacon.py | 98 ++- headroom/telemetry/session.py | 1056 +++++++++++++++++++++++++++++++ tests/conftest.py | 15 + tests/test_telemetry_warning.py | 52 ++ 11 files changed, 1855 insertions(+), 13 deletions(-) create mode 100644 deploy/beacon/.gitignore create mode 100755 deploy/beacon/query.sh create mode 100644 deploy/beacon/sample-event.json create mode 100644 deploy/beacon/worker.js create mode 100644 deploy/beacon/wrangler.toml create mode 100644 headroom/telemetry/session.py diff --git a/deploy/beacon/.gitignore b/deploy/beacon/.gitignore new file mode 100644 index 000000000..6abe1d268 --- /dev/null +++ b/deploy/beacon/.gitignore @@ -0,0 +1,2 @@ +# wrangler local state, caches, and account info — never commit +.wrangler/ diff --git a/deploy/beacon/query.sh b/deploy/beacon/query.sh new file mode 100755 index 000000000..9ca5f3d18 --- /dev/null +++ b/deploy/beacon/query.sh @@ -0,0 +1,94 @@ +#!/usr/bin/env bash +# Query the telemetry corpus in R2 with DuckDB. +# +# ./query.sh # fleet summary +# ./query.sh sessions # one row per session (deduped) +# ./query.sh "SELECT ..." # your own SQL against the corpus +# +# Setup, once: +# brew install duckdb +# Cloudflare > R2 > API > Create Account API Token (Object Read only, +# scoped to headroom-telemetry), then put the values in ~/env.txt +# (or any file named by HEADROOM_ENV_FILE): +# +# R2_ACCOUNT_ID=... +# R2_ACCESS_KEY_ID=... +# R2_SECRET_ACCESS_KEY=... +# +# R2_ACCOUNT_TOKEN is Cloudflare's REST-API token and is NOT used here — the +# S3 protocol wants the access-key pair. +set -euo pipefail + +BUCKET="${R2_BUCKET:-headroom-telemetry}" +_repo_env="$(cd "$(dirname "${BASH_SOURCE[0]}")/../.." && pwd)/.env" +ENV_FILE="${HEADROOM_ENV_FILE:-$HOME/env.txt}" +[ -f "$ENV_FILE" ] || ENV_FILE="$_repo_env" + +[ -f "$ENV_FILE" ] || { echo "no env file (~/env.txt or $_repo_env) — see this script's header" >&2; exit 1; } +# shellcheck disable=SC1090 +set -a; source "$ENV_FILE"; set +a + +for v in R2_ACCOUNT_ID R2_ACCESS_KEY_ID R2_SECRET_ACCESS_KEY; do + [ -n "${!v:-}" ] || { echo "$v not set in $ENV_FILE" >&2; exit 1; } +done +command -v duckdb >/dev/null || { echo "duckdb not installed: brew install duckdb" >&2; exit 1; } + +# Credentials go in via a heredoc on stdin, never on the command line, so they +# stay out of `ps` and shell history. +SECRET=" +INSTALL httpfs; LOAD httpfs; +CREATE OR REPLACE SECRET r2corpus ( + TYPE r2, + KEY_ID '${R2_ACCESS_KEY_ID}', + SECRET '${R2_SECRET_ACCESS_KEY}', + ACCOUNT_ID '${R2_ACCOUNT_ID}' +); +" + +# The corpus is heartbeats: a session reports every 5 minutes with CUMULATIVE +# totals under one id. So the row with the highest seq per (install, session) is +# the whole session — never SUM across heartbeats, you would count each session +# once per report. +DEDUPE=" +CREATE OR REPLACE TEMP VIEW sessions AS +SELECT * FROM read_ndjson('r2://${BUCKET}/sessions/**/*.json', union_by_name = true) +QUALIFY row_number() OVER ( + PARTITION BY resource['headroom.install_id'], session.id + ORDER BY session.seq DESC +) = 1; +" + +case "${1:-summary}" in + summary) + # Fleet rates come from summing raw counts. Averaging the per-session + # rates.*_pct fields would weight a 10-token session equal to a 1M one. + QUERY=" + SELECT count(*) AS sessions, + count(DISTINCT resource['headroom.install_id']) AS installs, + sum(session.turns) AS turns, + sum(tokens.saved) AS tokens_saved, + sum(tokens.tool_saved) AS tool_tokens_saved, + round(sum(tokens.attempted) * 100.0 + / nullif(sum(tokens.original), 0), 2) AS eligible_pct, + round(sum(tokens.saved) * 100.0 + / nullif(sum(tokens.attempted), 0), 2) AS yield_pct, + round(sum(tokens.saved) * 100.0 + / nullif(sum(tokens.original), 0), 2) AS saved_pct, + sum(failures) AS failures + FROM sessions;" + ;; + sessions) + QUERY=" + SELECT resource['headroom.install_id'][1:8] AS install, + session.id, session.seq, session.turns, session.duration_s, + tokens.original, tokens.attempted, tokens.saved, + rates.saved_pct, rates.eligible_pct, rates.yield_pct, + providers, models, skips + FROM sessions + ORDER BY session.duration_s DESC + LIMIT 50;" + ;; + *) QUERY="$1" ;; +esac + +printf '%s\n%s\n%s\n' "$SECRET" "$DEDUPE" "$QUERY" | duckdb -box diff --git a/deploy/beacon/sample-event.json b/deploy/beacon/sample-event.json new file mode 100644 index 000000000..6d5e13bd5 --- /dev/null +++ b/deploy/beacon/sample-event.json @@ -0,0 +1,297 @@ +{ + "resourceLogs": [ + { + "resource": { + "attributes": [ + { + "key": "service.name", + "value": { + "stringValue": "headroom" + } + }, + { + "key": "service.version", + "value": { + "stringValue": "0.34.0" + } + }, + { + "key": "headroom.install_id", + "value": { + "stringValue": "00000000000000000000000000000000" + } + }, + { + "key": "os.type", + "value": { + "stringValue": "darwin" + } + }, + { + "key": "host.arch", + "value": { + "stringValue": "arm64" + } + } + ] + }, + "scopeLogs": [ + { + "scope": { + "name": "headroom.telemetry.session" + }, + "logRecords": [ + { + "timeUnixNano": "1785731364402434048", + "body": { + "kvlistValue": { + "values": [ + { + "key": "schema_version", + "value": { + "intValue": "1" + } + }, + { + "key": "session", + "value": { + "kvlistValue": { + "values": [ + { + "key": "id", + "value": { + "stringValue": "sample0000000001" + } + }, + { + "key": "seq", + "value": { + "intValue": "0" + } + }, + { + "key": "duration_s", + "value": { + "intValue": "4210" + } + }, + { + "key": "turns", + "value": { + "intValue": "47" + } + }, + { + "key": "ended", + "value": { + "stringValue": "active" + } + }, + { + "key": "final", + "value": { + "boolValue": false + } + } + ] + } + } + }, + { + "key": "tokens", + "value": { + "kvlistValue": { + "values": [ + { + "key": "original", + "value": { + "intValue": "890000" + } + }, + { + "key": "attempted", + "value": { + "intValue": "410000" + } + }, + { + "key": "input", + "value": { + "intValue": "570000" + } + }, + { + "key": "output", + "value": { + "intValue": "41000" + } + }, + { + "key": "saved", + "value": { + "intValue": "320000" + } + }, + { + "key": "tool_saved", + "value": { + "intValue": "48000" + } + }, + { + "key": "cache_read", + "value": { + "intValue": "210000" + } + }, + { + "key": "cache_write", + "value": { + "intValue": "30000" + } + }, + { + "key": "uncached", + "value": { + "intValue": "650000" + } + } + ] + } + } + }, + { + "key": "rates", + "value": { + "kvlistValue": { + "values": [ + { + "key": "saved_pct", + "value": { + "doubleValue": 35.96 + } + }, + { + "key": "eligible_pct", + "value": { + "doubleValue": 46.07 + } + }, + { + "key": "yield_pct", + "value": { + "doubleValue": 78.05 + } + }, + { + "key": "cache_read_pct", + "value": { + "doubleValue": 23.6 + } + }, + { + "key": "overhead_pct", + "value": { + "doubleValue": 1.96 + } + } + ] + } + } + }, + { + "key": "compression", + "value": { + "kvlistValue": { + "values": [ + { + "key": "transforms", + "value": { + "kvlistValue": { + "values": [ + { + "key": "crush", + "value": { + "intValue": "47" + } + } + ] + } + } + }, + { + "key": "overhead_ms_total", + "value": { + "intValue": "1840" + } + }, + { + "key": "latency_ms_total", + "value": { + "intValue": "94000" + } + }, + { + "key": "passthrough_turns", + "value": { + "intValue": "0" + } + }, + { + "key": "response_cache_hits", + "value": { + "intValue": "3" + } + } + ] + } + } + }, + { + "key": "skips", + "value": { + "kvlistValue": { + "values": [] + } + } + }, + { + "key": "providers", + "value": { + "arrayValue": { + "values": [ + { + "stringValue": "anthropic" + } + ] + } + } + }, + { + "key": "models", + "value": { + "arrayValue": { + "values": [ + { + "stringValue": "claude-sonnet-4-5-20250929" + } + ] + } + } + }, + { + "key": "failures", + "value": { + "intValue": "2" + } + } + ] + } + } + } + ] + } + ] + } + ] +} diff --git a/deploy/beacon/worker.js b/deploy/beacon/worker.js new file mode 100644 index 000000000..d4e2d0ac2 --- /dev/null +++ b/deploy/beacon/worker.js @@ -0,0 +1,174 @@ +/** + * Headroom telemetry beacon receiver. + * + * This file is open source on purpose. It is the other half of the promise + * made in headroom/telemetry/session.py: users can read exactly what the + * client sends AND exactly what happens to it on arrival. "Trust us" is not a + * privacy policy. + * + * Deployed at otlp.headroomlabs.ai. Three jobs: + * + * 1. Allowlist. Drop every field not on ALLOWED_KEYS before anything is + * written. This is the only privacy control that works retroactively — + * if a future client version ships a bug that leaks a field, we cannot + * patch the installs already in the wild, but we can stop storing it + * here in one deploy. + * + * 2. Flatten. OTLP AnyValue nesting is portable but miserable to query + * ({"kvlistValue":{"values":[{"key":"tokens",...}]}}). We keep OTLP on + * the wire so the backend stays vendor-swappable, and store plain JSON so + * DuckDB can read it without unwrapping anything. + * + * 3. Fan out. R2 for the durable corpus; optionally a metrics vendor for + * dashboards. Adding a destination is one more call here — never a + * client release. + * + * What this deliberately does NOT do: log, store, or forward the source IP. + * Cloudflare offers it as cf-connecting-ip; it is the one field that would + * deanonymise install_id, so it is never read. + */ + +// Mirrors the payload built by _Session.payload(). A key absent here is +// dropped, not stored. Adding a metric means adding it here first — that +// friction is the point. +const ALLOWED_KEYS = [ + 'schema_version', + 'session', + 'tokens', + 'rates', + 'compression', + 'skips', + 'sources', + 'providers', + 'models', + 'failures', +]; + +// Resource attributes we keep. Same rule: allowlist, not denylist. +const ALLOWED_RESOURCE = [ + 'service.name', + 'service.version', + 'headroom.install_id', + 'headroom.install_mode', + 'headroom.stack', + 'os.type', + 'host.arch', +]; + +// A beacon event is ~2KB. Anything far past that is a bug or an attack. +const MAX_BODY_BYTES = 64 * 1024; + +/** OTLP AnyValue -> plain JS. The inverse of _any_value() in session.py. */ +function unwrap(value) { + if (value == null) return null; + if ('stringValue' in value) return value.stringValue; + if ('boolValue' in value) return value.boolValue; + if ('intValue' in value) return Number(value.intValue); + if ('doubleValue' in value) return value.doubleValue; + if ('arrayValue' in value) return (value.arrayValue.values || []).map(unwrap); + if ('kvlistValue' in value) { + const out = {}; + for (const kv of value.kvlistValue.values || []) out[kv.key] = unwrap(kv.value); + return out; + } + return null; +} + +function pick(obj, allowed) { + const out = {}; + if (!obj || typeof obj !== 'object') return out; + for (const key of allowed) { + if (key in obj) out[key] = obj[key]; + } + return out; +} + +/** OTLP ExportLogsServiceRequest -> flat, allowlisted records. */ +function extract(payload) { + const records = []; + for (const rl of payload.resourceLogs || []) { + const resource = {}; + for (const attr of rl.resource?.attributes || []) { + resource[attr.key] = unwrap(attr.value); + } + const cleanResource = pick(resource, ALLOWED_RESOURCE); + + for (const sl of rl.scopeLogs || []) { + for (const rec of sl.logRecords || []) { + const body = unwrap(rec.body); + if (!body || typeof body !== 'object') continue; + records.push({ + ...pick(body, ALLOWED_KEYS), + resource: cleanResource, + // Server-stamped. A client clock can be wrong or forged; this is the + // timestamp partitioning and retention actually rely on. + received_at: new Date().toISOString(), + }); + } + } + } + return records; +} + +export default { + async fetch(request, env, ctx) { + if (request.method !== 'POST') { + return new Response('beacon: POST OTLP logs to /v1/logs', { status: 405 }); + } + const url = new URL(request.url); + if (url.pathname !== '/v1/logs') { + return new Response('not found', { status: 404 }); + } + + const raw = await request.arrayBuffer(); + if (raw.byteLength > MAX_BODY_BYTES) { + return new Response('payload too large', { status: 413 }); + } + + let records; + try { + records = extract(JSON.parse(new TextDecoder().decode(raw))); + } catch { + // Malformed input is not worth a retry storm from clients. + return new Response('bad request', { status: 400 }); + } + if (records.length === 0) return new Response(null, { status: 204 }); + + const now = new Date(); + const day = now.toISOString().slice(0, 10); + const hour = now.toISOString().slice(11, 13); + // Hive-style partitioning so DuckDB can prune by date without a catalog. + // ponytail: one object per request. At beacon volume that is a few hundred + // thousand objects a month, which globs fine. Add a daily compaction job + // when the file count starts to slow queries, not before. + const key = `sessions/dt=${day}/hh=${hour}/${crypto.randomUUID()}.json`; + const ndjson = records.map((r) => JSON.stringify(r)).join('\n'); + + // Respond immediately; durability work continues after the response. + // The client is fire-and-forget and ignores the status anyway — making it + // wait on R2 would only add latency to someone else's coding session. + ctx.waitUntil( + env.CORPUS.put(key, ndjson, { + httpMetadata: { contentType: 'application/x-ndjson' }, + }) + ); + + // Optional second lane: forward verbatim OTLP to a metrics backend for + // dashboards. Configured by secret, so it can be added or swapped with a + // `wrangler secret put` and no code change. + if (env.METRICS_OTLP_URL) { + ctx.waitUntil( + fetch(env.METRICS_OTLP_URL, { + method: 'POST', + headers: { + 'content-type': 'application/json', + authorization: env.METRICS_OTLP_AUTH || '', + }, + body: JSON.stringify({ resourceLogs: [{ scopeLogs: [{ logRecords: records.map((r) => ({ body: { stringValue: JSON.stringify(r) } })) }] }] }), + }).catch(() => {}) + ); + } + + return new Response(null, { status: 204 }); + }, +}; diff --git a/deploy/beacon/wrangler.toml b/deploy/beacon/wrangler.toml new file mode 100644 index 000000000..412b22045 --- /dev/null +++ b/deploy/beacon/wrangler.toml @@ -0,0 +1,44 @@ +name = "headroom-beacon" +main = "worker.js" +compatibility_date = "2025-01-01" + +# PHASE 1 — deploy to ..workers.dev with no DNS changes. +# Lets the whole path be tested against a real client before headroomlabs.ai +# nameservers move anywhere. +workers_dev = true + +# PHASE 2 — the permanent address. Uncomment once headroomlabs.ai is on +# Cloudflare nameservers, then redeploy. This string is baked into every +# released client (DEFAULT_ENDPOINT in headroom/telemetry/session.py), so it can +# never change afterwards — everything behind it can. +# +# Deploying this while the zone is still on Namecheap fails: wrangler cannot +# find the zone. That is the intended guardrail, not a bug. +# +# [[routes]] +# pattern = "otlp.headroomlabs.ai/v1/logs" +# zone_name = "headroomlabs.ai" +# custom_domain = false + +# The corpus. R2 rather than S3 specifically for zero egress: training jobs +# re-read the whole dataset, and on S3 that is a recurring bill for data we +# already own. +[[r2_buckets]] +binding = "CORPUS" +bucket_name = "headroom-telemetry" + +# Optional metrics lane, added later without touching this file: +# npx wrangler secret put METRICS_OTLP_URL +# npx wrangler secret put METRICS_OTLP_AUTH +# Absent = R2 only, which is the right place to start. + +[observability] +enabled = true + +# Rate limiting is configured in the Cloudflare dashboard, not here — this +# endpoint is unauthenticated by design (anonymity is the product), so it is +# the only thing between the Worker and a bored stranger: +# Security > WAF > Rate limiting rules +# otlp.headroomlabs.ai/v1/logs -> 60 requests / minute / IP +# A real client sends ~2 requests/hour, so that is ~1000x headroom while still +# capping a single abusive source hard. diff --git a/headroom/ccr/mcp_server.py b/headroom/ccr/mcp_server.py index 9fa2b69ad..18a75f3d7 100644 --- a/headroom/ccr/mcp_server.py +++ b/headroom/ccr/mcp_server.py @@ -36,6 +36,7 @@ from typing import Any from headroom import paths as _paths from headroom import savings_ledger from headroom.cache.compression_store import format_retrieval_miss_detail +from headroom.telemetry import session as telemetry_session # fcntl is Unix-only; on Windows we skip file locking (stats are best-effort). # Keep the module typed as Any so Windows mypy runs don't try to resolve Unix-only attrs. @@ -795,6 +796,16 @@ class HeadroomMCPServer: client=self._current_client(), source="mcp", ) + # Anonymous beacon (opt-out; no-op unless enabled). Without this the MCP + # path is invisible to aggregate stats even though it is a first-class + # way to use Headroom — and subagents make that worse, since each runs + # its own MCP server. Swallows its own errors; the caller's try/except + # is a second net, not the first. + telemetry_session.record_mcp_compression( + original_tokens=before, + compressed_tokens=after, + model=os.environ.get("HEADROOM_MCP_MODEL"), + ) def _current_client(self) -> str: """Name of the MCP client driving this session (best-effort).""" diff --git a/headroom/proxy/outcome.py b/headroom/proxy/outcome.py index e789d452a..3c92b4c9b 100644 --- a/headroom/proxy/outcome.py +++ b/headroom/proxy/outcome.py @@ -314,8 +314,11 @@ class RequestOutcome: async def emit_request_outcome(handler: Any, outcome: RequestOutcome) -> None: """Single funnel for per-request bookkeeping. The contract. - Owns the four downstream effects in canonical order: + Owns the downstream effects in canonical order: + 0. ``telemetry.session.record_outcome(...)`` — anonymous session beacon + (opt-in, no-op unless ``HEADROOM_TELEMETRY`` is on). Runs ahead of the + 5xx guard below so session error rates see upstream failures. 1. ``handler.metrics.record_request(...)`` — Prometheus / SavingsTracker 2. ``handler.cost_tracker.record_tokens(...)`` — cost dashboard (skipped when cost_tracker is None, i.e. ``--no-cost``) @@ -324,8 +327,8 @@ async def emit_request_outcome(handler: Any, outcome: RequestOutcome) -> None: 4. structured PERF log line — consumed by ``headroom perf`` A failure outcome (``status_code >= 500``, e.g. a 529 surfaced after retry - exhaustion) short-circuits before these four effects: it records a failed - request and returns, so an upstream failure cannot feed the success stats. + exhaustion) short-circuits before effects 1-4: it records a failed request + and returns, so an upstream failure cannot feed the success stats. Takes the handler as a free argument rather than ``self`` so this function is callable from: @@ -343,6 +346,7 @@ async def emit_request_outcome(handler: Any, outcome: RequestOutcome) -> None: from headroom.proxy.cost import _summarize_transforms from headroom.proxy.models import RequestLog from headroom.proxy.project_context import get_current_project + from headroom.telemetry.session import record_outcome # GitHub Copilot: requests routed to the Copilot API travel on the OpenAI or # Anthropic wire, so the handlers stamp the wire provider. Relabel to @@ -357,6 +361,21 @@ async def emit_request_outcome(handler: Any, outcome: RequestOutcome) -> None: outcome = dataclasses.replace(outcome, provider="copilot") + # 0. Anonymous session beacon. Opt-in and a no-op unless HEADROOM_TELEMETRY + # is explicitly on, in which case it folds this outcome into an in-memory + # per-session aggregate and POSTs one content-free event when the session + # goes idle (off-thread — see telemetry.session.post_session_event). + # + # Placed before the 5xx short-circuit, for the same reason the Copilot + # relabel above is: a session's failure count has to see upstream + # failures, and per-session error rate is exactly the signal that shows a + # provider going flaky for real users. Everything below this point is + # success-only bookkeeping. + # + # Synchronous but allocation-light, and swallows its own exceptions — the + # beacon must never add latency to, or take down, the request path. + record_outcome(outcome) + # Upstream failure (>= 500, e.g. a 529 Overloaded surfaced after retry # exhaustion) must not feed the savings/cost/log success stats; that would # let a failed request inflate the save-rate. Record it as failed and stop, diff --git a/headroom/telemetry/beacon.py b/headroom/telemetry/beacon.py index c94272a21..eda40869e 100644 --- a/headroom/telemetry/beacon.py +++ b/headroom/telemetry/beacon.py @@ -1,14 +1,23 @@ -"""Telemetry opt-in state for Headroom. +"""Telemetry state for Headroom. -Headroom collects only **local**, aggregate telemetry (tokens saved, compression -ratios, timings) for the in-process collector and the ``/stats`` / -``/v1/telemetry`` endpoints. **Nothing is sent to Headroom Labs:** the anonymous -telemetry beacon that previously shipped aggregate stats has been removed. -Operational metrics can still be exported to *your own* OpenTelemetry collector -via ``HEADROOM_OTEL_METRICS_*`` (a destination you control). +Two independent switches, because they answer two different questions: -This module holds the ``HEADROOM_TELEMETRY`` opt-in predicate (off by default) -that gates local collection, plus the CLI notice helpers. +``HEADROOM_TELEMETRY`` — **off by default, opt-in.** Aggregates stats locally +for the in-process collector and the ``/stats`` / ``/v1/telemetry`` endpoints. +Never leaves the machine. + +``HEADROOM_BEACON`` — **on by default, opt-out.** Uploads an anonymous, +content-free session summary to Headroom Labs. Disable with +``HEADROOM_BEACON=off``, ``DO_NOT_TRACK=1``, or offline mode. See +:mod:`headroom.telemetry.session` for the exact payload and +``deploy/beacon/worker.js`` for what the receiver does with it. + +Operational metrics are separate again, and go to *your own* OpenTelemetry +collector via ``HEADROOM_OTEL_METRICS_*`` — a destination you control, that +Headroom Labs never sees. + +Keeping these separate matters: an operator who turned on local stats has not +thereby agreed to upload anything, and must not start doing so on upgrade. """ from __future__ import annotations @@ -36,6 +45,53 @@ def is_telemetry_enabled() -> bool: return val in _ON_VALUES +# Whether the upload beacon runs when the operator has expressed no preference. +# +# True makes Headroom opt-OUT: every install uploads anonymous session summaries +# unless it is told not to. Flip to False for opt-in and nothing uploads without +# an explicit HEADROOM_BEACON=on. +# +# This single constant is the whole policy — deliberately, so the decision is +# one line to audit and one line to reverse. +BEACON_DEFAULT_ON = True + + +def is_beacon_enabled() -> bool: + """Check if the anonymous upload beacon is enabled. + + Default is :data:`BEACON_DEFAULT_ON`. Three things switch it off regardless, + in this order: + + * offline mode — no network is no network; + * ``DO_NOT_TRACK`` — the cross-tool convention, honoured because a user who + has already stated this preference should not have to state it again; + * ``HEADROOM_BEACON=off`` (or false/0/no/disable/disabled). + + Deliberately a *separate* switch from ``HEADROOM_TELEMETRY``. That flag + means "aggregate stats locally so ``/stats`` has something to show" — data + that never leaves the machine. Overloading it to also authorise upload + would mean every operator who had turned on local stats begins transmitting + the moment they upgrade, having answered a different question. + + Note the asymmetry with :func:`is_telemetry_enabled`, which is fail-closed: + this one is fail-open by construction while ``BEACON_DEFAULT_ON`` is True, + so an unrecognised value uploads rather than staying silent. + """ + from headroom.offline import is_offline + + if is_offline(): + return False + dnt = os.environ.get("DO_NOT_TRACK", "").lower().strip() + if dnt and dnt not in _OFF_VALUES: + return False + raw = os.environ.get("HEADROOM_BEACON", "").lower().strip() + if raw in _OFF_VALUES: + return False + if raw in _ON_VALUES: + return True + return BEACON_DEFAULT_ON + + def is_telemetry_warn_enabled() -> bool: """Check if telemetry warnings are enabled (feature flag, on by default). @@ -56,8 +112,30 @@ def format_telemetry_notice(*, prefix: str = "") -> str: Returns an empty string when telemetry or warnings are disabled so callers can unconditionally include the result in their output. """ - if not is_telemetry_enabled() or not is_telemetry_warn_enabled(): + if not is_telemetry_warn_enabled(): return "" + + beacon = is_beacon_enabled() + local = is_telemetry_enabled() + if not beacon and not local: + return "" + + # The beacon line comes first and is unconditional when on. It is opt-out, + # so this notice is the only place a user learns it is running at all — + # surprise is what turns anonymous telemetry into a trust incident. + # + # Wording: lead with what the data *is* (compression ratios) and why it is + # useful (understanding when compression works), not with the fact of + # transmission. "Stats are sent to Headroom Labs" is accurate but reads like + # extraction, which misrepresents a payload that is entirely counters. The + # reassurance has to stay concrete, though — "never prompts, code, or file + # paths" names the things people actually worry about, and every one of + # those is enforced by the allowlist, not just promised here. + if beacon: + return ( + f"{prefix}Telemetry: anonymous compression stats — never prompts, code, or " + "file paths. Helps us improve compression | Off: HEADROOM_BEACON=off" + ) return ( f"{prefix}Telemetry: ENABLED (local aggregate stats only — nothing sent externally) | " "Disable: HEADROOM_TELEMETRY=off or --no-telemetry" diff --git a/headroom/telemetry/session.py b/headroom/telemetry/session.py new file mode 100644 index 000000000..e7d9f0933 --- /dev/null +++ b/headroom/telemetry/session.py @@ -0,0 +1,1056 @@ +"""Session aggregation for the anonymous telemetry beacon. + +One wide event per session, emitted when the session goes idle. Everything here +is off unless ``HEADROOM_TELEMETRY`` is explicitly on — see +:mod:`headroom.telemetry.beacon`. + +# What counts as a session + +A contiguous burst of proxy activity from one install, closed after +``IDLE_TIMEOUT_S`` of quiet. This is the web-analytics definition (GA uses a +30-minute inactivity window) and it is deliberately *not* keyed on conversation +content. + +The alternative was to key sessions on +:func:`headroom.proxy.output_savings_policy.conversation_key_from_body`, which +is stable across the turns of one agent loop. That was rejected twice over: + +* It is content-derived — a SHA256 over the first 512 chars of the first user + message. Exporting it, even blinded, puts a function of the user's prompt on + the wire. For a proxy whose entire promise is safe prompt handling, that is + the wrong default, and defending it in a threat model costs more than the + grouping is worth. +* ``RequestOutcome`` is built at 30 sites across six handlers, and the parsed + body is not available at any shared chokepoint (the HTTP middleware sees + headers only — see ``set_current_project`` at ``proxy/server.py``). Plumbing + a conversation key to all of them is a large diff for a metric nobody has + asked for yet. + +Burst sessions serve every metric the beacon exists for: session count, +duration, turns, tokens saved per session, and retention. What they lose is +conversation *boundaries* — two concurrent conversations merge into one burst. +``turn_id`` is already on every outcome if conversation-level grouping is ever +needed; it can be added without changing this shape. + +# Wire format + +OTLP/HTTP JSON logs, POSTed straight to the collector. Deliberately *not* the +OTel logs SDK: that API is still ``opentelemetry.sdk._logs`` (underscore = +unstable), and the beacon should not break on an SDK minor bump. The OTLP JSON +wire format is a stable spec and plain stdlib gets us there. + +The record body is an OTLP kvlist (not a JSON string) so the collector's +``keep_keys(body, [...])`` allowlist can introspect and drop unknown fields +server-side. That is the only control point for clients already in the wild. +""" + +from __future__ import annotations + +import atexit +import json +import logging +import os +import platform +import re +import secrets +import threading +import time +import urllib.request +import uuid +from collections.abc import Callable +from dataclasses import dataclass, field +from pathlib import Path +from typing import Any + +logger = logging.getLogger(__name__) + +# Where session events go. The receiver is open source: deploy/beacon/worker.js. +# +# TEMPORARY: this is a Cloudflare workers.dev address, which is a *vendor* +# hostname. OSS users pin versions and old releases live for years, so whatever +# ships in a tagged release is effectively permanent — a vendor URL here marries +# Headroom to Cloudflare and looks alarming to anyone who runs `strings` on the +# package. Move this to otlp.headroomlabs.ai (one CNAME, or a Cloudflare zone) +# before cutting a release that contains it. +DEFAULT_ENDPOINT = "https://headroom-beacon.headroom-beacon.workers.dev/v1/logs" +SCHEMA_VERSION = 1 + +# A gap longer than this closes the session. Agent loops are bursty; 15 minutes +# separates "reading the diff before the next prompt" from "walked away". +IDLE_TIMEOUT_S = 900.0 + +# How often a live session reports in. Each report is a *cumulative snapshot* +# of the same session, not a slice of it — see _Session.payload. +FLUSH_INTERVAL_S = 300.0 + +_POST_TIMEOUT_S = 5.0 + +# Reason/enum values are bounded vocabularies in code, but they reach us as +# free strings. Validate rather than trust: anything off-pattern becomes +# "other" so a future tag value cannot smuggle text onto the wire. +_SLUG_RE = re.compile(r"^[a-z][a-z0-9_]{0,31}$") + +# Tags whose values are skip/bypass reasons — the "why was compression low" +# vocabulary. Values are slug-validated before they are counted. +_REASON_TAGS = ("passthrough_reason", "image_skip_reason", "memory_skip_reason") + + +def _pct(numerator: float, denominator: float) -> float: + """Percentage to 2dp, or 0.0 when undefined. + + Zero denominator returns 0.0 rather than null: the raw counts ship + alongside every ratio, so "0.0 with original=0" is unambiguous, and it + keeps the field a plain number for every consumer. + """ + if not denominator: + return 0.0 + return round(numerator / denominator * 100.0, 2) + + +def _safe_slug(value: Any) -> str: + text = str(value or "").strip().lower() + return text if _SLUG_RE.match(text) else "other" + + +_model_cache: dict[str, str | None] = {} + + +def _public_model(model: str) -> str | None: + """Return the model id only if it appears in a public model registry. + + Model ids are public product SKUs, not user data — withholding them costs + real signal (Haiku and Opus are the same ``provider`` but completely + different compression economics). The exception is custom deployments: + ``ft:gpt-4o:acme-corp:internal-bot:abc123`` carries an org name, and a + self-hosted model can be called anything at all. + + litellm's cost map is exactly the "is this a public SKU" oracle, and + Headroom already consults it for pricing in ``proxy/savings_tracker.py``. + Unknown model, or litellm absent (it is gated to Python < 3.14), means no + model field. Under-reporting is the correct failure direction here. + """ + if not model: + return None + if model in _model_cache: + return _model_cache[model] + + resolved: str | None = None + try: + import litellm + + registry = litellm.model_cost + for candidate in (model, model.rsplit("/", 1)[-1], model.rsplit(".", 1)[-1]): + if candidate in registry: + resolved = candidate + break + except Exception: + resolved = None + + # ponytail: unbounded dict, but it is keyed by distinct model ids seen in + # one process — a handful. Cap it if a router ever fans out over hundreds. + _model_cache[model] = resolved + return resolved + + +# -------------------------------------------------------------------------- +# install identity +# -------------------------------------------------------------------------- + +_identity_lock = threading.Lock() +_install_id: str | None = None + + +def _install_id_path() -> Path: + from headroom.paths import config_dir + + return config_dir() / "install_id" + + +def install_id() -> str: + """Random per-install id. Not a machine fingerprint — delete the file to reset. + + Deliberately not derived from hostname, MAC, or any hardware property: + anything a user cannot change reads as tracking, and a random UUID answers + every question the beacon actually asks. + """ + global _install_id + with _identity_lock: + if _install_id is not None: + return _install_id + path = _install_id_path() + try: + existing = path.read_text().strip() + if existing: + _install_id = existing + return _install_id + except OSError: + pass + _install_id = uuid.uuid4().hex + try: + path.parent.mkdir(parents=True, exist_ok=True) + fd = os.open(path, os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600) + with os.fdopen(fd, "w") as fh: + fh.write(_install_id) + except OSError: + logger.debug("telemetry: could not persist install_id", exc_info=True) + return _install_id + + +def resource_attributes( + *, install_mode: str | None = None, stack: str | None = None +) -> dict[str, str]: + """Set once per process; every signal inherits these for free. + + Adding a field here applies it to every signal, retroactively — this is the + cheap lane. ``install_mode`` and ``stack`` are passed in rather than + detected: :func:`headroom.telemetry.context.detect_install_mode` needs the + bound port and :func:`~headroom.telemetry.context.detect_stack` needs the + live stats dict, both of which only the proxy has. + """ + from headroom._version import get_version + + attrs = { + "service.name": "headroom", + "service.version": get_version(), + "headroom.install_id": install_id(), + "os.type": platform.system().lower(), + "host.arch": platform.machine().lower(), + } + if install_mode: + attrs["headroom.install_mode"] = install_mode + if stack: + attrs["headroom.stack"] = stack + return attrs + + +# -------------------------------------------------------------------------- +# aggregation +# -------------------------------------------------------------------------- + + +@dataclass +class _Session: + sid: str + started: float + last_seen: float + last_emit: float = 0.0 + seq: int = 0 + turns: int = 0 + original_tokens: int = 0 + attempted_tokens: int = 0 + input_tokens: int = 0 + output_tokens: int = 0 + tokens_saved: int = 0 + tool_saved_tokens: int = 0 + cache_read_tokens: int = 0 + cache_write_tokens: int = 0 + uncached_tokens: int = 0 + failures: int = 0 + passthrough_turns: int = 0 + response_cache_hits: int = 0 + overhead_ms: float = 0.0 + latency_ms: float = 0.0 + transforms: dict[str, int] = field(default_factory=dict) + skips: dict[str, int] = field(default_factory=dict) + sources: dict[str, int] = field(default_factory=dict) + providers: set[str] = field(default_factory=set) + models: set[str] = field(default_factory=set) + + def payload(self, reason: str) -> dict[str, Any]: + """Cumulative snapshot of this session. Content-free. + + Every counter is a running total since the session began, NOT a delta + since the last report. That is what makes the stream dedupable: a + session that reports 8 times produces 8 rows sharing one ``id``, and + the one with the highest ``seq`` is the complete picture. Deduping is + then a window function, and no summing is involved: + + SELECT * FROM events + QUALIFY row_number() OVER ( + PARTITION BY resource['headroom.install_id'], session.id + ORDER BY session.seq DESC + ) = 1 + + Cumulative also makes the stream loss-tolerant: a dropped report costs + nothing, because the next one restates everything. Deltas would leave a + permanent hole. The redundancy is ~2KB per report, which is free. + + ``seq`` increments on every emission, so this is not a pure read. + """ + snapshot = { + "schema_version": SCHEMA_VERSION, + "session": { + "id": self.sid, + "seq": self.seq, + "duration_s": int(self.last_seen - self.started), + "turns": self.turns, + # "active" means more reports are coming for this id. Anything + # else is the last word on this session. + "ended": reason, + "final": reason != "active", + }, + # original -> attempted -> saved is the whole diagnostic chain. + # A low `saved` means two completely different things: + # attempted << original -> little was eligible (frozen cache + # prefix, passthrough, system prompts) + # saved << attempted -> compression ran and found nothing, + # which is the real quality signal + # Shipping only `saved` cannot tell those apart. + "tokens": { + "original": self.original_tokens, + "attempted": self.attempted_tokens, + "input": self.input_tokens, + "output": self.output_tokens, + "saved": self.tokens_saved, + # Tool-schema savings (deferral + turn-hook shrink) never move + # original/optimized — outcome.py says so explicitly. Without + # this field a tool-heavy session reports saved=0 while having + # genuinely saved thousands, which understates the product. + "tool_saved": self.tool_saved_tokens, + "cache_read": self.cache_read_tokens, + "cache_write": self.cache_write_tokens, + "uncached": self.uncached_tokens, + }, + # Convenience ratios for this ONE session, 0.0 when the denominator + # is zero. Do not average these across sessions to get a fleet + # number — that weights a 10-token session equal to a 1M-token one. + # Fleet rates come from summing the raw counts above. + "rates": { + # Of everything sent, how much did we remove? + "saved_pct": _pct(self.tokens_saved, self.original_tokens), + # Of everything sent, how much were we even allowed to touch? + # A low value here means frozen cache prefix / passthrough / + # system prompts — a ceiling, not a compressor failure. + "eligible_pct": _pct(self.attempted_tokens, self.original_tokens), + # Of what we touched, how much did we remove? THIS is the + # compressor quality number, and the one Kompress moves. + "yield_pct": _pct(self.tokens_saved, self.attempted_tokens), + # Provider prompt cache participation. Headroom freezes prefixes + # to protect this, so it is the other side of eligible_pct. + "cache_read_pct": _pct(self.cache_read_tokens, self.original_tokens), + # What fraction of wall-clock did Headroom itself add? + "overhead_pct": _pct(self.overhead_ms, self.latency_ms), + }, + "compression": { + "transforms": dict(self.transforms), + "overhead_ms_total": round(self.overhead_ms, 1), + "latency_ms_total": round(self.latency_ms, 1), + "passthrough_turns": self.passthrough_turns, + # Served from Headroom's own response cache — the provider was + # never called at all. 100% saving on those turns. + "response_cache_hits": self.response_cache_hits, + }, + # Why compression did not fire, by bounded reason slug. + "skips": dict(self.skips), + # Turns by origin: "proxy" for requests through the HTTP proxy, + # "mcp" for headroom_compress tool calls. Those have different + # shapes — an MCP call has no provider, no upstream latency, and + # everything handed to it is eligible — so aggregate stats must be + # able to separate them rather than silently blending the two. + "sources": dict(self.sources), + "providers": sorted(self.providers), + "models": sorted(self.models), + "failures": self.failures, + } + self.seq += 1 + return snapshot + + +class SessionAggregator: + """Folds per-request outcomes into one event per activity burst. + + Thread-safe. One live session per process, so there is no LRU to bound and + nothing to evict. Reaping happens on write rather than on a background + task — an idle process has nothing to flush anyway, and ``flush_all`` + covers shutdown. + """ + + def __init__( + self, + emit: Callable[[dict[str, Any]], None], + *, + idle_s: float = IDLE_TIMEOUT_S, + flush_s: float = FLUSH_INTERVAL_S, + ) -> None: + self._emit = emit + self._idle_s = idle_s + self._flush_s = flush_s + self._lock = threading.Lock() + self._current: _Session | None = None + + def record(self, outcome: Any, *, source: str = "proxy", now: float | None = None) -> None: + """Fold one request outcome into the live session. Never raises.""" + now = time.time() if now is None else now + pending: list[dict[str, Any]] = [] + try: + with self._lock: + live = self._current + # Quiet for longer than the idle window: that session is over. + # We only notice on the next request, so its closing report is + # late — but every counter in it is already correct, because + # snapshots are cumulative. + if live is not None and now - live.last_seen >= self._idle_s: + pending.append(live.payload("idle")) + live = None + if live is None: + live = _Session( + sid=secrets.token_hex(8), + started=now, + last_seen=now, + last_emit=now, + ) + self._current = live + _fold(live, outcome, now, source) + # Heartbeat. Same session id, running totals — a later report + # supersedes an earlier one rather than adding to it. + if now - live.last_emit >= self._flush_s: + live.last_emit = now + pending.append(live.payload("active")) + except Exception: # telemetry must never break the proxy + logger.debug("telemetry: session record failed", exc_info=True) + return + # Emit outside the lock: the POST must not queue every other caller. + for snapshot in pending: + self._safe_emit(snapshot) + + def flush_all( + self, + reason: str = "shutdown", + *, + emit: Callable[[dict[str, Any]], None] | None = None, + ) -> None: + """Close and report the live session. + + ``emit`` overrides the configured sink for this call only. The shutdown + path needs that: the normal sink hands off to a daemon thread, and + daemon threads are killed before they finish once atexit handlers are + running, so the final report would never leave the process. + """ + with self._lock: + pending, self._current = self._current, None + if pending is None: + return + sink = emit or self._emit + try: + sink(pending.payload(reason)) + except Exception: + logger.debug("telemetry: session emit failed", exc_info=True) + + def _safe_emit(self, payload: dict[str, Any]) -> None: + try: + self._emit(payload) + except Exception: + logger.debug("telemetry: session emit failed", exc_info=True) + + +def _fold(sess: _Session, outcome: Any, now: float, source: str = "proxy") -> None: + """Duck-typed against RequestOutcome so field moves don't break the beacon. + + Duck-typing is what lets the MCP path reuse this: ``_McpCompression`` sets + only the handful of fields it actually knows, and everything else falls + through to a neutral default instead of needing a fake RequestOutcome. + """ + + def get(name: str, default: Any = 0) -> Any: + return getattr(outcome, name, default) + + sess.last_seen = now + sess.turns += 1 + sess.sources[source] = sess.sources.get(source, 0) + 1 + sess.original_tokens += int(get("original_tokens") or 0) + sess.attempted_tokens += int(get("attempted_input_tokens") or 0) + sess.input_tokens += int(get("optimized_tokens") or 0) + sess.output_tokens += int(get("output_tokens") or 0) + sess.tokens_saved += int(get("tokens_saved") or 0) + sess.cache_read_tokens += int(get("cache_read_tokens") or 0) + sess.cache_write_tokens += int(get("cache_write_tokens") or 0) + sess.uncached_tokens += int(get("uncached_input_tokens") or 0) + sess.overhead_ms += float(get("overhead_ms", 0.0) or 0.0) + sess.latency_ms += float(get("total_latency_ms", 0.0) or 0.0) + if int(get("status_code", 200) or 200) >= 500: + sess.failures += 1 + if get("from_response_cache", False): + sess.response_cache_hits += 1 + + provider = get("provider", "") or "" + if provider: + sess.providers.add(str(provider)[:32]) + + public = _public_model(str(get("model", "") or "")) + if public: + sess.models.add(public) + + # Skip/bypass reasons. These are the answer to "why was compression low" — + # a session where every turn is passthrough:bypass_header is a config + # problem, not a compression problem, and the two are indistinguishable + # from the token counts alone. + tags = get("tags", None) or {} + if isinstance(tags, dict): + # Tool-schema savings live only in tags — see the same arithmetic in + # emit_request_outcome, which feeds the local dashboard. + for tag in ("tool_search_deferred_tokens", "turn_hook_tools_saved_tokens"): + try: + sess.tool_saved_tokens += int(tags.get(tag, 0) or 0) + except (TypeError, ValueError): + pass + for tag in _REASON_TAGS: + if tag in tags: + key = f"{tag.removesuffix('_reason')}:{_safe_slug(tags[tag])}" + sess.skips[key] = sess.skips.get(key, 0) + 1 + if "passthrough_reason" in tags: + sess.passthrough_turns += 1 + + for name in get("transforms_applied", ()) or (): + # Keep only the part before the first colon. Some transforms encode + # per-request detail in a suffix — output_shaper ships + # "output_shaper:stratum:||" + # (see output_savings_policy.stratum_label), which would put the model + # tier and a request-size bucket on the wire as a side effect of a + # label format that has nothing to do with telemetry. The beacon only + # wants "which transforms ran, how often"; the A/B detail belongs to + # the local recorder that owns it. + key = str(name).split(":", 1)[0][:64] + sess.transforms[key] = sess.transforms.get(key, 0) + 1 + + +# -------------------------------------------------------------------------- +# OTLP/HTTP JSON transport +# -------------------------------------------------------------------------- + + +def _any_value(value: Any) -> dict[str, Any]: + """Python -> OTLP AnyValue. bool is checked before int: bool subclasses int.""" + # OTLP has no null. Without this, None falls through to str() and ships the + # literal text "None" as a value. + if value is None: + return {} + if isinstance(value, bool): + return {"boolValue": value} + if isinstance(value, int): + return {"intValue": str(value)} # int64 is a string in OTLP JSON + if isinstance(value, float): + return {"doubleValue": value} + if isinstance(value, str): + return {"stringValue": value} + if isinstance(value, dict): + return { + "kvlistValue": { + "values": [{"key": str(k), "value": _any_value(v)} for k, v in value.items()] + } + } + if isinstance(value, (list, tuple, set)): + return {"arrayValue": {"values": [_any_value(v) for v in value]}} + return {"stringValue": str(value)} + + +def build_otlp_logs(payload: dict[str, Any], resource: dict[str, str]) -> dict[str, Any]: + """Wrap one payload in an OTLP/HTTP JSON ExportLogsServiceRequest.""" + return { + "resourceLogs": [ + { + "resource": { + "attributes": [ + {"key": k, "value": {"stringValue": v}} for k, v in resource.items() + ] + }, + "scopeLogs": [ + { + "scope": {"name": "headroom.telemetry.session"}, + "logRecords": [ + { + "timeUnixNano": str(int(time.time() * 1_000_000_000)), + "body": _any_value(payload), + } + ], + } + ], + } + ] + } + + +def _post_blocking(payload: dict[str, Any], timeout: float = _POST_TIMEOUT_S) -> None: + endpoint = os.environ.get("HEADROOM_TELEMETRY_ENDPOINT", DEFAULT_ENDPOINT) + try: + from headroom._version import get_version + + data = json.dumps( + build_otlp_logs(payload, resource_attributes()), separators=(",", ":") + ).encode() + request = urllib.request.Request( + endpoint, + data=data, + method="POST", + headers={ + "content-type": "application/json", + # Required, not cosmetic. urllib defaults to "Python-urllib/3.x", + # which Cloudflare blocks outright by browser signature (error + # 1010) before the request ever reaches the Worker. Combined + # with fire-and-forget error handling that failure mode is + # invisible: every upload 403s and the beacon looks healthy. + "user-agent": f"headroom-beacon/{get_version()}", + }, + ) + with urllib.request.urlopen(request, timeout=timeout): + pass + except Exception: + logger.debug("telemetry: session POST failed", exc_info=True) + + +def post_session_event(payload: dict[str, Any]) -> None: + """Ship one session event on a daemon thread. Never blocks, never raises. + + The caller is the proxy's outcome funnel, which is async — a synchronous + urllib POST there would stall the event loop for up to ``_POST_TIMEOUT_S``. + Sessions close at most once per ``IDLE_TIMEOUT_S``, so a thread per event + is a handful per hour; a pool would be machinery for nothing. + ponytail: unbounded thread spawn is safe only because the close rate is + bounded by the idle timeout. Revisit if anything else starts emitting here. + """ + try: + threading.Thread( + target=_post_blocking, + args=(payload,), + name="headroom-beacon", + daemon=True, + ).start() + except Exception: + logger.debug("telemetry: could not start beacon thread", exc_info=True) + + +_aggregator: SessionAggregator | None = None +_aggregator_lock = threading.Lock() + + +def get_session_aggregator() -> SessionAggregator: + global _aggregator + with _aggregator_lock: + if _aggregator is None: + _aggregator = SessionAggregator(post_session_event) + atexit.register(_flush_at_exit) + return _aggregator + + +# A shutdown POST must not hang the process. Shorter than the normal timeout: +# a user quitting their agent should not wait on our collector. +_EXIT_POST_TIMEOUT_S = 2.0 + + +def _flush_at_exit() -> None: + """Report the open session on graceful exit, synchronously. + + Synchronous on purpose. ``post_session_event`` normally hands off to a + daemon thread so the request path never blocks, but by the time atexit + handlers run the interpreter is shutting down and daemon threads are killed + before they can finish — a thread started here simply never posts. + + This is load-bearing well beyond the ``final`` marker: sessions shorter than + ``FLUSH_INTERVAL_S`` have not heartbeated even once, so without a working + exit flush every short session would report nothing at all. + + Still does not cover SIGKILL or a closed laptop. Heartbeats bound the loss + there to one flush interval. + """ + aggregator = _aggregator + if aggregator is None: + return + aggregator.flush_all(emit=lambda payload: _post_blocking(payload, timeout=_EXIT_POST_TIMEOUT_S)) + + +@dataclass(frozen=True) +class _McpCompression: + """Minimal outcome shim for a ``headroom_compress`` MCP tool call. + + Only the fields an MCP compression actually knows. Everything else _fold + reads falls through to a neutral default — there is no provider, no + upstream latency, and no cache participation, because the tool never talks + to a model. + + ``attempted_input_tokens == original_tokens`` on purpose: the caller hands + the tool exactly the content it wants compressed, so all of it is eligible. + That makes ``eligible_pct`` 100% for MCP turns, which is correct rather + than flattering — and it is why ``sources`` has to stay in the payload, so + MCP turns can be excluded when reading the proxy's eligibility ceiling. + """ + + original_tokens: int + attempted_input_tokens: int + optimized_tokens: int + tokens_saved: int + model: str = "" + + +def record_mcp_compression( + *, original_tokens: int, compressed_tokens: int, model: str | None = None +) -> None: + """Beacon entry point for the MCP tool path. + + Separate from :func:`record_outcome` because MCP servers are separate, + often short-lived processes — the main one plus one per subagent, per + ``headroom.savings_ledger``. Each gets its own aggregator and reports its + own session, which is accurate: that really is separate work. They share + ``install_id``, so the sessions can be grouped per install downstream. + + Short-lived processes depend entirely on the atexit flush, since they may + never live long enough to heartbeat. See :func:`_flush_at_exit`. + """ + from headroom.telemetry.beacon import is_beacon_enabled + + if not is_beacon_enabled(): + return + try: + before = int(original_tokens or 0) + after = int(compressed_tokens or 0) + except (TypeError, ValueError): + return + if before <= 0: + return + get_session_aggregator().record( + _McpCompression( + original_tokens=before, + attempted_input_tokens=before, + optimized_tokens=after, + tokens_saved=max(before - after, 0), + model=str(model or ""), + ), + source="mcp", + ) + + +def record_outcome(outcome: Any) -> None: + """Beacon entry point, called from the proxy's outcome funnel. + + Off by default and cheap when off: the enabled check short-circuits before + the aggregator is ever constructed, so a user who never opted in pays one + env lookup per request and allocates nothing. + """ + from headroom.telemetry.beacon import is_beacon_enabled + + if not is_beacon_enabled(): + return + get_session_aggregator().record(outcome) + + +def demo() -> None: + """Self-check: python -m headroom.telemetry.session""" + + class FakeOutcome: + provider = "anthropic" + model = "claude-3-5-sonnet-20241022" + optimized_tokens = 100 + output_tokens = 20 + tokens_saved = 50 + cache_read_tokens = 10 + overhead_ms = 1.5 + status_code = 200 + # Widened so subclasses below can override with a different arity — + # RequestOutcome declares tuple[str, ...] too. + transforms_applied: tuple[str, ...] = ("crush", "dedupe") + + # bool must not be encoded as int (bool subclasses int). + assert _any_value(True) == {"boolValue": True} + assert _any_value(1) == {"intValue": "1"} + assert _any_value(2.5) == {"doubleValue": 2.5} + assert _any_value({"a": [1, 2]})["kvlistValue"]["values"][0]["key"] == "a" + assert _any_value(None) == {}, "None must not serialise as the text 'None'" + + # Ratios: the three the product is judged on, plus a zero-denominator guard. + assert _pct(25, 100) == 25.0 + assert _pct(1, 3) == 33.33 + assert _pct(5, 0) == 0.0 + + class Rich(FakeOutcome): + original_tokens = 1000 + attempted_input_tokens = 400 # 40% eligible + tokens_saved = 300 # 30% overall, 75% yield on eligible + cache_read_tokens = 500 # 50% cache read + cache_write_tokens = 100 + uncached_input_tokens = 400 + total_latency_ms = 1000.0 + overhead_ms = 50.0 # 5% overhead + from_response_cache = True + tags = {"tool_search_deferred_tokens": "800", "turn_hook_tools_saved_tokens": 200} + + rates_out: list[dict[str, Any]] = [] + ra = SessionAggregator(rates_out.append) + ra.record(Rich(), now=100.0) + ra.flush_all() + r = rates_out[0] + assert r["rates"]["saved_pct"] == 30.0, r["rates"] + assert r["rates"]["eligible_pct"] == 40.0, r["rates"] + assert r["rates"]["yield_pct"] == 75.0, r["rates"] + assert r["rates"]["cache_read_pct"] == 50.0, r["rates"] + assert r["rates"]["overhead_pct"] == 5.0, r["rates"] + # Tool savings are invisible in `saved` by design; they must not be lost. + assert r["tokens"]["tool_saved"] == 1000, r["tokens"] + assert r["compression"]["response_cache_hits"] == 1 + assert r["tokens"]["cache_write"] == 100 and r["tokens"]["uncached"] == 400 + + emitted: list[dict[str, Any]] = [] + agg = SessionAggregator(emitted.append, idle_s=10.0) + + agg.record(FakeOutcome(), now=1000.0) + agg.record(FakeOutcome(), now=1002.0) + assert emitted == [], "a live session must not emit" + + # Quiet past the timeout, then activity -> the old burst closes. + agg.record(FakeOutcome(), now=1100.0) + assert len(emitted) == 1, emitted + event = emitted[0] + assert event["session"]["turns"] == 2 + assert event["session"]["ended"] == "idle" + assert event["session"]["duration_s"] == 2 + assert event["tokens"]["saved"] == 100 + assert event["tokens"]["input"] == 200 + assert event["compression"]["transforms"] == {"crush": 2, "dedupe": 2} + assert event["providers"] == ["anthropic"] + assert event["failures"] == 0 + + # The new burst is a distinct session, not a continuation. + agg.flush_all() + assert len(emitted) == 2, emitted + assert emitted[1]["session"]["turns"] == 1 + assert emitted[1]["session"]["id"] != emitted[0]["session"]["id"] + assert emitted[1]["session"]["ended"] == "shutdown" + + # Nothing derived from prompt content or the model id reaches the wire. + wire = json.dumps(emitted) + assert "sonnet" not in wire and "claude" not in wire, wire + + # Model ids reach the wire only when a public registry knows them. A + # fine-tune id carries an org name and must never survive. + assert _public_model("ft:gpt-4o:acme-corp:internal-bot:abc123") is None + assert _public_model("acme-internal-llama") is None + assert _public_model("") is None + if _public_model("gpt-4o") is None: + print(" (litellm unavailable — model field degrades to absent)") + else: + assert _public_model("gpt-4o") == "gpt-4o" + + # Reason slugs are validated, not trusted. + assert _safe_slug("bypass_header") == "bypass_header" + assert _safe_slug("/Users/me/secret/path.py") == "other" + assert _safe_slug(None) == "other" + + # Low compression must be explainable: a passthrough session and a + # ran-but-found-nothing session must not look the same. + class Bypassed(FakeOutcome): + original_tokens = 5000 + attempted_input_tokens = 0 + tokens_saved = 0 + transforms_applied = () + tags = {"passthrough_reason": "bypass_header"} + + class Barren(FakeOutcome): + original_tokens = 5000 + attempted_input_tokens = 4000 + tokens_saved = 12 + tags: dict[str, str] = {} + + diag: list[dict[str, Any]] = [] + agg_bypassed = SessionAggregator(diag.append) + agg_bypassed.record(Bypassed(), now=5000.0) + agg_bypassed.flush_all() + agg_barren = SessionAggregator(diag.append) + agg_barren.record(Barren(), now=6000.0) + agg_barren.flush_all() + + bypassed, barren = diag[0], diag[1] + assert bypassed["tokens"]["attempted"] == 0 + assert bypassed["skips"] == {"passthrough:bypass_header": 1} + assert bypassed["compression"]["passthrough_turns"] == 1 + assert barren["tokens"]["attempted"] == 4000 + assert barren["skips"] == {} + # Both saved ~nothing, but the reason is now distinguishable. + assert bypassed["tokens"]["saved"] == 0 and barren["tokens"]["saved"] == 12 + + # A live session heartbeats under ONE id with cumulative totals, so the + # highest-seq row is the whole session and dedupe is last-write-wins. + beats: list[dict[str, Any]] = [] + hb = SessionAggregator(beats.append, idle_s=900.0, flush_s=100.0) + hb.record(FakeOutcome(), now=7000.0) + hb.record(FakeOutcome(), now=7050.0) + assert beats == [], "must not emit before the flush interval" + + hb.record(FakeOutcome(), now=7100.0) # first heartbeat + hb.record(FakeOutcome(), now=7150.0) + hb.record(FakeOutcome(), now=7250.0) # second heartbeat + hb.flush_all() # final + + assert len(beats) == 3, beats + ids = {b["session"]["id"] for b in beats} + assert len(ids) == 1, f"one session must not fragment into {len(ids)} ids" + assert [b["session"]["seq"] for b in beats] == [0, 1, 2] + assert [b["session"]["final"] for b in beats] == [False, False, True] + assert [b["session"]["ended"] for b in beats] == ["active", "active", "shutdown"] + + # Cumulative, not deltas: turns and tokens only ever climb, and the last + # row alone reconstructs the session. + assert [b["session"]["turns"] for b in beats] == [3, 5, 5] + assert [b["tokens"]["saved"] for b in beats] == [150, 250, 250] + + # Dedupe by (install, session id), keep max seq -> exactly one row. + latest: dict[str, dict[str, Any]] = {} + for b in beats: + key = b["session"]["id"] + if key not in latest or b["session"]["seq"] > latest[key]["session"]["seq"]: + latest[key] = b + assert len(latest) == 1 + only = next(iter(latest.values())) + assert only["session"]["turns"] == 5 and only["tokens"]["saved"] == 250 + + # Losing a heartbeat costs nothing — the survivor still restates everything. + survivors = [beats[0], beats[2]] + assert max(s["session"]["seq"] for s in survivors) == 2 + assert survivors[-1]["session"]["turns"] == 5 + + # Going quiet starts a genuinely new session, not a continuation. + hb.record(FakeOutcome(), now=90000.0) + hb.flush_all() + assert beats[-1]["session"]["id"] != beats[0]["session"]["id"] + assert beats[-1]["session"]["turns"] == 1 + + # A transform label carrying a stratum suffix must be reduced to its prefix + # — otherwise output_shaper smuggles the model tier and a size bucket out. + class Shaped(FakeOutcome): + transforms_applied = ("output_shaper:stratum:sonnet|tool_result|8k", "crush") + + shaped: list[dict[str, Any]] = [] + agg_s = SessionAggregator(shaped.append) + agg_s.record(Shaped(), now=1500.0) + agg_s.flush_all() + assert shaped[0]["compression"]["transforms"] == {"output_shaper": 1, "crush": 1} + assert "sonnet" not in json.dumps(shaped[0]), shaped[0] + + # A 5xx counts as a failure without poisoning the token stats. + class Failed(FakeOutcome): + status_code = 529 + + agg2 = SessionAggregator(emitted.append) + agg2.record(Failed(), now=2000.0) + agg2.flush_all() + assert emitted[-1]["failures"] == 1 + + # Flushing an empty aggregator is a no-op, not a null event. + before = len(emitted) + SessionAggregator(emitted.append).flush_all() + assert len(emitted) == before + + # MCP turns must be distinguishable from proxy turns in the same session. + # Blending them would corrupt eligible_pct: everything handed to the tool is + # eligible by construction, so MCP turns always read 100% and would drag the + # proxy's real eligibility ceiling upward. + mixed: list[dict[str, Any]] = [] + mx = SessionAggregator(mixed.append) + mx.record(FakeOutcome(), now=9000.0) + mx.record( + _McpCompression( + original_tokens=1000, + attempted_input_tokens=1000, + optimized_tokens=300, + tokens_saved=700, + ), + source="mcp", + now=9001.0, + ) + mx.flush_all() + mixed_ev = mixed[0] + assert mixed_ev["sources"] == {"proxy": 1, "mcp": 1}, mixed_ev["sources"] + assert mixed_ev["session"]["turns"] == 2 + # An MCP-only shim contributes tokens but no provider and no latency. + assert mixed_ev["tokens"]["saved"] == 700 + 50 + assert mixed_ev["providers"] == ["anthropic"] # only the proxy turn had one + + mcp_only: list[dict[str, Any]] = [] + mo = SessionAggregator(mcp_only.append) + mo.record( + _McpCompression( + original_tokens=800, + attempted_input_tokens=800, + optimized_tokens=200, + tokens_saved=600, + ), + source="mcp", + now=9100.0, + ) + mo.flush_all() + only_ev = mcp_only[0] + assert only_ev["sources"] == {"mcp": 1} + assert only_ev["rates"]["eligible_pct"] == 100.0 + assert only_ev["rates"]["yield_pct"] == 75.0 + assert only_ev["rates"]["saved_pct"] == 75.0 + assert only_ev["providers"] == [] and only_ev["models"] == [] + # No upstream call means no latency, and the ratio must not divide by zero. + assert only_ev["rates"]["overhead_pct"] == 0.0 + + # record_mcp_compression: a real call records, a degenerate one does not. + # Beacon forced on for this block so the assertions cannot pass merely + # because the ambient environment has it disabled. + _prev_agg = _aggregator + _prev_env = { + k: os.environ.get(k) for k in ("HEADROOM_BEACON", "DO_NOT_TRACK", "HEADROOM_OFFLINE") + } + os.environ["HEADROOM_BEACON"] = "on" + # DO_NOT_TRACK and offline mode outrank an explicit opt-in, so they have to + # be cleared here or these assertions test the wrong thing. + os.environ.pop("DO_NOT_TRACK", None) + os.environ.pop("HEADROOM_OFFLINE", None) + try: + globals()["_aggregator"] = SessionAggregator(lambda _p: None) + record_mcp_compression(original_tokens=0, compressed_tokens=0) + assert globals()["_aggregator"]._current is None, "zero-token call recorded" + + record_mcp_compression(original_tokens=500, compressed_tokens=100) + live = globals()["_aggregator"]._current + assert live is not None, "valid MCP call did not record" + assert live.sources == {"mcp": 1} + assert live.tokens_saved == 400 + + # A compression that grew the content must not report negative savings. + record_mcp_compression(original_tokens=100, compressed_tokens=250) + assert globals()["_aggregator"]._current.tokens_saved == 400 + + # DO_NOT_TRACK outranks an explicit HEADROOM_BEACON=on. + os.environ["DO_NOT_TRACK"] = "1" + globals()["_aggregator"] = SessionAggregator(lambda _p: None) + record_mcp_compression(original_tokens=500, compressed_tokens=100) + assert globals()["_aggregator"]._current is None, "DO_NOT_TRACK was ignored" + os.environ.pop("DO_NOT_TRACK", None) + finally: + globals()["_aggregator"] = _prev_agg + for _k, _v in _prev_env.items(): + if _v is None: + os.environ.pop(_k, None) + else: + os.environ[_k] = _v + + # flush_all must honour an emit override. The atexit path depends on this: + # the normal sink defers to a daemon thread, and daemon threads are killed + # before they finish once the interpreter is shutting down, so a thread + # started there never posts. Without the override, every session shorter + # than FLUSH_INTERVAL_S would report nothing at all. + default_sink: list[dict[str, Any]] = [] + override_sink: list[dict[str, Any]] = [] + ex = SessionAggregator(default_sink.append) + ex.record(FakeOutcome(), now=8000.0) + ex.flush_all(emit=override_sink.append) + assert override_sink and not default_sink, "flush_all ignored the emit override" + assert override_sink[0]["session"]["final"] is True + + # An emit that blows up must not propagate to the caller. + def boom(_payload: dict[str, Any]) -> None: + raise RuntimeError("collector down") + + agg3 = SessionAggregator(boom, idle_s=1.0) + agg3.record(FakeOutcome(), now=3000.0) + agg3.record(FakeOutcome(), now=3100.0) + agg3.flush_all() + + # A malformed outcome must not raise either. + SessionAggregator(emitted.append).record(object(), now=4000.0) + + print("ok") + + +if __name__ == "__main__": + demo() diff --git a/tests/conftest.py b/tests/conftest.py index 38de51499..2aecdbc4a 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -30,6 +30,21 @@ def _scrub_developer_headroom_env(monkeypatch): monkeypatch.delenv("ANTHROPIC_CUSTOM_HEADERS", raising=False) +# The scrub above deletes every HEADROOM_* var — which includes HEADROOM_BEACON, +# and the beacon defaults to ON. So scrubbing for hermeticity is precisely what +# switches it on, and with HEADROOM_TELEMETRY_ENDPOINT scrubbed too it falls back +# to the real production endpoint. Every test that reaches the outcome funnel +# then POSTs a session event for real: observed writing into the live corpus +# during a local run, and CI would do the same on every push. +# +# Depends on the scrub fixture so it is guaranteed to run after it rather than +# relying on declaration order. A test that wants the beacon on just sets the +# var itself — monkeypatch inside the test wins over this. +@pytest.fixture(autouse=True) +def _disable_telemetry_beacon(monkeypatch, _scrub_developer_headroom_env): + monkeypatch.setenv("HEADROOM_BEACON", "off") + + # The MCP install ledger defaults to ``~/.headroom/mcp_installs.json``, so any # test that registers a server (directly or through `wrap`) writes into the # developer's REAL ledger — observed adding a live `claude/serena` entry during a diff --git a/tests/test_telemetry_warning.py b/tests/test_telemetry_warning.py index dbd5a7f9f..c00b8bf1a 100644 --- a/tests/test_telemetry_warning.py +++ b/tests/test_telemetry_warning.py @@ -53,6 +53,7 @@ class TestFormatTelemetryNotice: def test_returns_notice_when_telemetry_on(self, monkeypatch): monkeypatch.setenv("HEADROOM_TELEMETRY", "on") + monkeypatch.setenv("HEADROOM_BEACON", "off") monkeypatch.delenv("HEADROOM_TELEMETRY_WARN", raising=False) notice = format_telemetry_notice() assert notice != "" @@ -62,6 +63,43 @@ class TestFormatTelemetryNotice: def test_empty_when_telemetry_off(self, monkeypatch): monkeypatch.setenv("HEADROOM_TELEMETRY", "off") + monkeypatch.setenv("HEADROOM_BEACON", "off") + monkeypatch.delenv("HEADROOM_TELEMETRY_WARN", raising=False) + assert format_telemetry_notice() == "" + + def test_beacon_is_announced_by_default(self, monkeypatch): + """The beacon is opt-out, so the notice is the only place a user finds + out it is running. Silence here is how anonymous telemetry becomes a + trust incident.""" + for var in ("HEADROOM_TELEMETRY", "HEADROOM_BEACON", "DO_NOT_TRACK"): + monkeypatch.delenv(var, raising=False) + monkeypatch.delenv("HEADROOM_TELEMETRY_WARN", raising=False) + notice = format_telemetry_notice() + assert "compression stats" in notice + assert "HEADROOM_BEACON=off" in notice + + def test_beacon_notice_names_what_is_not_sent(self, monkeypatch): + """Vague reassurance is worse than none. The notice has to name the + three things users actually worry about.""" + for var in ("HEADROOM_TELEMETRY", "HEADROOM_BEACON", "DO_NOT_TRACK"): + monkeypatch.delenv(var, raising=False) + monkeypatch.delenv("HEADROOM_TELEMETRY_WARN", raising=False) + notice = format_telemetry_notice() + assert "never prompts" in notice + assert "code" in notice + assert "file paths" in notice + + def test_silent_when_beacon_disabled_and_no_local(self, monkeypatch): + monkeypatch.setenv("HEADROOM_BEACON", "off") + for var in ("HEADROOM_TELEMETRY", "DO_NOT_TRACK"): + monkeypatch.delenv(var, raising=False) + monkeypatch.delenv("HEADROOM_TELEMETRY_WARN", raising=False) + assert format_telemetry_notice() == "" + + def test_do_not_track_silences_the_beacon_notice(self, monkeypatch): + monkeypatch.setenv("DO_NOT_TRACK", "1") + for var in ("HEADROOM_TELEMETRY", "HEADROOM_BEACON"): + monkeypatch.delenv(var, raising=False) monkeypatch.delenv("HEADROOM_TELEMETRY_WARN", raising=False) assert format_telemetry_notice() == "" @@ -176,6 +214,7 @@ class TestWrapCLITelemetryNotice: def test_print_notice_outputs_when_telemetry_on(self, monkeypatch, capsys): monkeypatch.setenv("HEADROOM_TELEMETRY", "on") + monkeypatch.setenv("HEADROOM_BEACON", "off") monkeypatch.delenv("HEADROOM_TELEMETRY_WARN", raising=False) from headroom.cli.wrap import _print_telemetry_notice @@ -185,8 +224,21 @@ class TestWrapCLITelemetryNotice: assert "Telemetry" in captured.out assert "HEADROOM_TELEMETRY=off" in captured.out + def test_print_notice_announces_beacon_by_default(self, monkeypatch, capsys): + for var in ("HEADROOM_TELEMETRY", "HEADROOM_BEACON", "DO_NOT_TRACK"): + monkeypatch.delenv(var, raising=False) + monkeypatch.delenv("HEADROOM_TELEMETRY_WARN", raising=False) + + from headroom.cli.wrap import _print_telemetry_notice + + _print_telemetry_notice() + captured = capsys.readouterr() + assert "compression stats" in captured.out + assert "HEADROOM_BEACON=off" in captured.out + def test_print_notice_silent_when_telemetry_off(self, monkeypatch, capsys): monkeypatch.setenv("HEADROOM_TELEMETRY", "off") + monkeypatch.setenv("HEADROOM_BEACON", "off") from headroom.cli.wrap import _print_telemetry_notice