feat: add anonymous telemetry warning across all user-facing paths

- Add is_telemetry_warn_enabled() feature flag (HEADROOM_TELEMETRY_WARN=off
  to suppress at build/pack time) and format_telemetry_notice() helper to beacon.py
- Export new helpers from headroom.telemetry.__init__
- Proxy CLI startup banner now shows Telemetry: ENABLED/DISABLED with opt-out
  instructions when enabled
- Log telemetry status at proxy server startup() so it appears in log stream
- wrap CLI _launch_tool() and the bespoke claude wrap command both call
  _print_telemetry_notice() so users see the notice before the tool launches
- /stats endpoint now includes anon_telemetry_shipping boolean flag
- Dashboard header shows amber Anon Telemetry indicator chip when
  anon_telemetry_shipping is true (theme-matching, with tooltip showing opt-out)
- 30 new tests covering all paths; all 82 tests pass

Agent-Logs-Url: https://github.com/JerrettDavis/headroom/sessions/880a7bb3-3ad9-49f4-a0b6-3ffdde233e48

Co-authored-by: JerrettDavis <2610199+JerrettDavis@users.noreply.github.com>

headroom/cli/proxy.py

@@ -333,6 +333,17 @@ Memory (Multi-Provider):
   - Database: {config.memory_db_path}
 """
 
+    from headroom.telemetry.beacon import is_telemetry_enabled
+
+    # Build telemetry section for the startup banner
+    if is_telemetry_enabled():
+        telemetry_line = (
+            "  Telemetry:    ENABLED (anonymous aggregate stats)\n"
+            "                Disable: HEADROOM_TELEMETRY=off  or  headroom proxy --no-telemetry"
+        )
+    else:
+        telemetry_line = "  Telemetry:    DISABLED"
+
     click.echo(f"""
 ╔═══════════════════════════════════════════════════════════════════════╗
 ║                         HEADROOM PROXY                                 ║
@@ -348,6 +359,7 @@ Starting proxy server...
   Rate Limit:   {"ENABLED" if config.rate_limit_enabled else "DISABLED"}
   Memory:       {memory_status}
   License:      {license_status}
+{telemetry_line}
 {backend_section}
 Routing:
   /v1/messages         → {anthropic_url}

headroom/cli/wrap.py

@@ -35,6 +35,20 @@ import click
 
 from .main import main
 
+
+def _print_telemetry_notice() -> None:
+    """Print a telemetry notice when anonymous telemetry is enabled.
+
+    Respects the HEADROOM_TELEMETRY and HEADROOM_TELEMETRY_WARN feature flags.
+    Does nothing when telemetry or warnings are disabled.
+    """
+    from headroom.telemetry.beacon import format_telemetry_notice
+
+    notice = format_telemetry_notice(prefix="  ")
+    if notice:
+        click.echo(notice)
+
+
 # Proxy health check (reused from evals/suite_runner.py pattern)
 
 
@@ -434,6 +448,7 @@ def _launch_tool(
             click.echo(f"  {var}")
         if args:
             click.echo(f"  Extra args: {' '.join(args)}")
+        _print_telemetry_notice()
         click.echo()
 
         result = subprocess.run([binary, *args], env=env)
@@ -735,6 +750,7 @@ def claude(
         click.echo(f"  ANTHROPIC_BASE_URL=http://127.0.0.1:{port}")
         if claude_args:
             click.echo(f"  Extra args: {' '.join(claude_args)}")
+        _print_telemetry_notice()
         click.echo()
 
         env = os.environ.copy()

headroom/dashboard/templates/dashboard.html

@@ -51,6 +51,18 @@
                     Historical
                 </button>
             </div>
+            <template x-if="stats.anon_telemetry_shipping">
+                <div class="inline-flex items-center gap-1.5 rounded-full border border-amber-500/40 bg-amber-500/10 px-2.5 py-1"
+                     title="Anonymous aggregate telemetry is enabled. Disable with HEADROOM_TELEMETRY=off or --no-telemetry.">
+                    <svg xmlns="http://www.w3.org/2000/svg" viewBox="0 0 16 16" fill="currentColor"
+                         class="w-3 h-3 text-amber-400 shrink-0">
+                        <path fill-rule="evenodd"
+                              d="M6.701 2.25c.577-1 2.02-1 2.598 0l5.196 9a1.5 1.5 0 0 1-1.299 2.25H2.804a1.5 1.5 0 0 1-1.3-2.25l5.197-9ZM8 4a.75.75 0 0 1 .75.75v3a.75.75 0 0 1-1.5 0v-3A.75.75 0 0 1 8 4Zm0 8a1 1 0 1 0 0-2 1 1 0 0 0 0 2Z"
+                              clip-rule="evenodd"/>
+                    </svg>
+                    <span class="text-xs text-amber-400 font-medium">Anon Telemetry</span>
+                </div>
+            </template>
             <div class="flex items-center gap-2">
                 <span class="text-xs text-gray-500">Status</span>
                 <span class="flex items-center gap-1.5">

headroom/proxy/server.py

@@ -127,6 +127,7 @@ from headroom.proxy.rate_limiter import TokenBucketRateLimiter  # noqa: F401
 from headroom.proxy.request_logger import RequestLogger  # noqa: F401
 from headroom.proxy.semantic_cache import SemanticCache  # noqa: F401
 from headroom.telemetry import get_telemetry_collector
+from headroom.telemetry.beacon import is_telemetry_enabled
 from headroom.telemetry.toin import get_toin
 from headroom.transforms import (
     CacheAligner,
@@ -616,6 +617,15 @@ class HeadroomProxy(
             logger.info("CCR: DISABLED")
         logger.info(f"Savings history: {self.metrics.savings_tracker.storage_path}")
 
+        # Log anonymous telemetry status so operators can see it in the log stream
+        if is_telemetry_enabled():
+            logger.info(
+                "Anonymous telemetry: ENABLED (aggregate stats only — no prompts or content). "
+                "Opt out: HEADROOM_TELEMETRY=off or --no-telemetry"
+            )
+        else:
+            logger.info("Anonymous telemetry: DISABLED")
+
     async def shutdown(self):
         """Cleanup async resources."""
         if self.http_client:
@@ -1169,6 +1179,7 @@ def create_app(config: ProxyConfig | None = None) -> FastAPI:
                 "ccr_retrievals": compression_stats.get("total_retrievals", 0),
             },
             "compression_cache": compression_cache_stats,
+            "anon_telemetry_shipping": is_telemetry_enabled(),
             "telemetry": {
                 "enabled": telemetry_stats.get("enabled", False),
                 "total_compressions": telemetry_stats.get("total_compressions", 0),

headroom/telemetry/__init__.py

@@ -47,6 +47,11 @@ TOIN (Tool Output Intelligence Network):
     toin.record_retrieval(sig_hash, retrieval_type, query, query_fields)
 """
 
+from .beacon import (
+    format_telemetry_notice,
+    is_telemetry_enabled,
+    is_telemetry_warn_enabled,
+)
 from .collector import (
     TelemetryCollector,
     TelemetryConfig,
@@ -70,6 +75,10 @@ from .toin import (
 )
 
 __all__ = [
+    # Beacon helpers
+    "format_telemetry_notice",
+    "is_telemetry_enabled",
+    "is_telemetry_warn_enabled",
     # Collector
     "TelemetryCollector",
     "TelemetryConfig",

headroom/telemetry/beacon.py

@@ -39,10 +39,41 @@ _ENDPOINT = f"{_SUPABASE_URL}/rest/v1/{_TABLE}?on_conflict=session_id"
 _INTERVAL_SECONDS = 300
 
 
+_OFF_VALUES = frozenset(("off", "false", "0", "no", "disable", "disabled"))
+
+
 def is_telemetry_enabled() -> bool:
     """Check if telemetry is enabled (on by default, opt out with env var)."""
     val = os.environ.get("HEADROOM_TELEMETRY", "on").lower().strip()
-    return val not in ("off", "false", "0", "no", "disable", "disabled")
+    return val not in _OFF_VALUES
+
+
+def is_telemetry_warn_enabled() -> bool:
+    """Check if telemetry warnings are enabled (feature flag, on by default).
+
+    Set HEADROOM_TELEMETRY_WARN=off to suppress startup/wrap notices.
+    This is a build/pack-time feature flag intended for operators who want
+    to disable the notice without disabling telemetry itself.
+    """
+    val = os.environ.get("HEADROOM_TELEMETRY_WARN", "on").lower().strip()
+    return val not in _OFF_VALUES
+
+
+def format_telemetry_notice(*, prefix: str = "") -> str:
+    """Return a single-line telemetry notice suitable for CLI output.
+
+    Args:
+        prefix: Optional leading whitespace / box-drawing prefix.
+
+    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():
+        return ""
+    return (
+        f"{prefix}Telemetry:    ENABLED (anonymous aggregate stats) | "
+        "Disable: HEADROOM_TELEMETRY=off or --no-telemetry"
+    )
 
 
 class TelemetryBeacon:

tests/test_telemetry_warning.py

@@ -0,0 +1,263 @@
+"""Tests for anonymous telemetry warning feature.
+
+Covers:
+- is_telemetry_warn_enabled() feature flag
+- format_telemetry_notice() helper
+- proxy CLI banner includes telemetry status
+- wrap CLI prints telemetry notice
+- /stats endpoint exposes anon_telemetry_shipping flag
+"""
+
+from unittest.mock import patch
+
+import pytest
+
+click = pytest.importorskip("click")
+from click.testing import CliRunner  # noqa: E402
+
+from headroom.telemetry.beacon import (  # noqa: E402
+    format_telemetry_notice,
+    is_telemetry_warn_enabled,
+)
+
+# ---------------------------------------------------------------------------
+# is_telemetry_warn_enabled
+# ---------------------------------------------------------------------------
+
+
+class TestIsTelemetryWarnEnabled:
+    """Tests for the HEADROOM_TELEMETRY_WARN feature flag."""
+
+    def test_enabled_by_default(self, monkeypatch):
+        monkeypatch.delenv("HEADROOM_TELEMETRY_WARN", raising=False)
+        assert is_telemetry_warn_enabled() is True
+
+    @pytest.mark.parametrize("value", ["off", "OFF", "false", "0", "no", "disable", "disabled"])
+    def test_disabled_by_env_var(self, monkeypatch, value):
+        monkeypatch.setenv("HEADROOM_TELEMETRY_WARN", value)
+        assert is_telemetry_warn_enabled() is False
+
+    @pytest.mark.parametrize("value", ["on", "ON", "1", "yes", "true"])
+    def test_enabled_by_truthy_env_var(self, monkeypatch, value):
+        monkeypatch.setenv("HEADROOM_TELEMETRY_WARN", value)
+        assert is_telemetry_warn_enabled() is True
+
+
+# ---------------------------------------------------------------------------
+# format_telemetry_notice
+# ---------------------------------------------------------------------------
+
+
+class TestFormatTelemetryNotice:
+    """Tests for format_telemetry_notice()."""
+
+    def test_returns_notice_when_telemetry_on(self, monkeypatch):
+        monkeypatch.setenv("HEADROOM_TELEMETRY", "on")
+        monkeypatch.delenv("HEADROOM_TELEMETRY_WARN", raising=False)
+        notice = format_telemetry_notice()
+        assert notice != ""
+        assert "ENABLED" in notice
+        assert "HEADROOM_TELEMETRY=off" in notice
+        assert "--no-telemetry" in notice
+
+    def test_empty_when_telemetry_off(self, monkeypatch):
+        monkeypatch.setenv("HEADROOM_TELEMETRY", "off")
+        monkeypatch.delenv("HEADROOM_TELEMETRY_WARN", raising=False)
+        assert format_telemetry_notice() == ""
+
+    def test_empty_when_warn_flag_off(self, monkeypatch):
+        monkeypatch.setenv("HEADROOM_TELEMETRY", "on")
+        monkeypatch.setenv("HEADROOM_TELEMETRY_WARN", "off")
+        assert format_telemetry_notice() == ""
+
+    def test_prefix_is_applied(self, monkeypatch):
+        monkeypatch.setenv("HEADROOM_TELEMETRY", "on")
+        monkeypatch.delenv("HEADROOM_TELEMETRY_WARN", raising=False)
+        notice = format_telemetry_notice(prefix="  ")
+        assert notice.startswith("  ")
+
+    def test_no_prefix_by_default(self, monkeypatch):
+        monkeypatch.setenv("HEADROOM_TELEMETRY", "on")
+        monkeypatch.delenv("HEADROOM_TELEMETRY_WARN", raising=False)
+        notice = format_telemetry_notice()
+        # Default prefix is "" so the string should start with "Telemetry"
+        assert notice.startswith("Telemetry:")
+
+
+# ---------------------------------------------------------------------------
+# proxy CLI banner
+# ---------------------------------------------------------------------------
+
+
+class TestProxyCLITelemetryBanner:
+    """Proxy CLI startup banner must include telemetry status."""
+
+    @pytest.fixture
+    def runner(self):
+        return CliRunner()
+
+    def test_banner_shows_telemetry_enabled(self, runner, monkeypatch):
+        monkeypatch.delenv("HEADROOM_TELEMETRY", raising=False)
+
+        from headroom.cli.main import main
+
+        with patch("headroom.proxy.server.run_server", side_effect=SystemExit(0)):
+            result = runner.invoke(main, ["proxy"])
+
+        assert "Telemetry:" in result.output
+        assert "ENABLED" in result.output
+
+    def test_banner_shows_telemetry_disabled(self, runner, monkeypatch):
+        monkeypatch.setenv("HEADROOM_TELEMETRY", "off")
+
+        from headroom.cli.main import main
+
+        with patch("headroom.proxy.server.run_server", side_effect=SystemExit(0)):
+            result = runner.invoke(main, ["proxy"])
+
+        assert "Telemetry:" in result.output
+        assert "DISABLED" in result.output
+
+    def test_no_telemetry_flag_disables(self, runner, monkeypatch):
+        monkeypatch.delenv("HEADROOM_TELEMETRY", raising=False)
+
+        from headroom.cli.main import main
+
+        with patch("headroom.proxy.server.run_server", side_effect=SystemExit(0)):
+            result = runner.invoke(main, ["proxy", "--no-telemetry"])
+
+        assert "Telemetry:" in result.output
+        assert "DISABLED" in result.output
+
+    def test_banner_shows_opt_out_instructions_when_enabled(self, runner, monkeypatch):
+        monkeypatch.delenv("HEADROOM_TELEMETRY", raising=False)
+
+        from headroom.cli.main import main
+
+        with patch("headroom.proxy.server.run_server", side_effect=SystemExit(0)):
+            result = runner.invoke(main, ["proxy"])
+
+        assert "HEADROOM_TELEMETRY=off" in result.output or "--no-telemetry" in result.output
+
+
+# ---------------------------------------------------------------------------
+# wrap CLI telemetry notice
+# ---------------------------------------------------------------------------
+
+
+class TestWrapCLITelemetryNotice:
+    """_print_telemetry_notice() is called from wrap commands."""
+
+    def test_print_notice_outputs_when_telemetry_on(self, monkeypatch, capsys):
+        monkeypatch.setenv("HEADROOM_TELEMETRY", "on")
+        monkeypatch.delenv("HEADROOM_TELEMETRY_WARN", raising=False)
+
+        from headroom.cli.wrap import _print_telemetry_notice
+
+        _print_telemetry_notice()
+        captured = capsys.readouterr()
+        assert "Telemetry" in captured.out
+        assert "HEADROOM_TELEMETRY=off" in captured.out
+
+    def test_print_notice_silent_when_telemetry_off(self, monkeypatch, capsys):
+        monkeypatch.setenv("HEADROOM_TELEMETRY", "off")
+
+        from headroom.cli.wrap import _print_telemetry_notice
+
+        _print_telemetry_notice()
+        captured = capsys.readouterr()
+        assert captured.out == ""
+
+    def test_print_notice_silent_when_warn_flag_off(self, monkeypatch, capsys):
+        monkeypatch.setenv("HEADROOM_TELEMETRY", "on")
+        monkeypatch.setenv("HEADROOM_TELEMETRY_WARN", "off")
+
+        from headroom.cli.wrap import _print_telemetry_notice
+
+        _print_telemetry_notice()
+        captured = capsys.readouterr()
+        assert captured.out == ""
+
+
+# ---------------------------------------------------------------------------
+# /stats endpoint – anon_telemetry_shipping flag
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.asyncio
+class TestStatsEndpointTelemetryFlag:
+    """The /stats endpoint must expose anon_telemetry_shipping."""
+
+    pytest.importorskip("fastapi")
+
+    async def test_stats_includes_anon_telemetry_shipping_true(self, monkeypatch):
+        monkeypatch.delenv("HEADROOM_TELEMETRY", raising=False)
+        from headroom.proxy.server import ProxyConfig, create_app
+
+        app = create_app(
+            ProxyConfig(
+                cache_enabled=False,
+                rate_limit_enabled=False,
+                cost_tracking_enabled=False,
+            )
+        )
+
+        from httpx import ASGITransport, AsyncClient
+
+        async with AsyncClient(
+            transport=ASGITransport(app=app), base_url="http://test"
+        ) as client:
+            resp = await client.get("/stats")
+
+        assert resp.status_code == 200
+        data = resp.json()
+        assert "anon_telemetry_shipping" in data
+        assert data["anon_telemetry_shipping"] is True
+
+    async def test_stats_includes_anon_telemetry_shipping_false(self, monkeypatch):
+        monkeypatch.setenv("HEADROOM_TELEMETRY", "off")
+        from headroom.proxy.server import ProxyConfig, create_app
+
+        app = create_app(
+            ProxyConfig(
+                cache_enabled=False,
+                rate_limit_enabled=False,
+                cost_tracking_enabled=False,
+            )
+        )
+
+        from httpx import ASGITransport, AsyncClient
+
+        async with AsyncClient(
+            transport=ASGITransport(app=app), base_url="http://test"
+        ) as client:
+            resp = await client.get("/stats")
+
+        assert resp.status_code == 200
+        data = resp.json()
+        assert "anon_telemetry_shipping" in data
+        assert data["anon_telemetry_shipping"] is False
+
+
+# ---------------------------------------------------------------------------
+# telemetry __init__ exports
+# ---------------------------------------------------------------------------
+
+
+class TestTelemetryModuleExports:
+    """New helpers must be exported from headroom.telemetry."""
+
+    def test_is_telemetry_warn_enabled_exported(self):
+        from headroom.telemetry import is_telemetry_warn_enabled as fn
+
+        assert callable(fn)
+
+    def test_is_telemetry_enabled_exported(self):
+        from headroom.telemetry import is_telemetry_enabled as fn
+
+        assert callable(fn)
+
+    def test_format_telemetry_notice_exported(self):
+        from headroom.telemetry import format_telemetry_notice as fn
+
+        assert callable(fn)