fix: address PR review feedback for HF user telemetry

- Replace silent `except Exception: pass` with proper logging and
narrowed catch (binascii.Error, json.JSONDecodeError, etc.)
- Add IS_HF_SPACES to hardware.py as source of truth for HF Spaces
detection, use consistently in analysis.py and ui/app.py
- Update privacy docs to reflect intentional raw username storage
- Add tracker-level, event dataclass, and edge case tests for new
is_logged_in/hf_username fields

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

src/mosaic/analysis.py

@@ -26,6 +26,7 @@ from mosaic.telemetry import extract_user_info
 # Import centralized hardware detection
 from mosaic.hardware import (
     spaces,
+    IS_HF_SPACES,
     IS_ZEROGPU,
     IS_T4_GPU,
     GPU_TYPE,
@@ -808,7 +809,7 @@ def analyze_slide(
         raise gr.Error("Please upload a slide.")
 
     # Extract user info for telemetry (HF Spaces only)
-    user_info = extract_user_info(request, IS_ZEROGPU)
+    user_info = extract_user_info(request, IS_HF_SPACES)
 
     # Initialize telemetry for resource tracking
     slide_start_time = time_module.time()

src/mosaic/hardware.py

@@ -8,6 +8,9 @@ import os
 import torch
 from loguru import logger
 
+# Detect HuggingFace Spaces environment
+IS_HF_SPACES = bool(os.environ.get("SPACE_ID"))
+
 # Detect HuggingFace Spaces ZeroGPU environment
 try:
     import spaces
@@ -95,6 +98,7 @@ def get_gpu_metrics() -> dict:
 __all__ = [
     "spaces",
     "HAS_SPACES",
+    "IS_HF_SPACES",
     "IS_ZEROGPU",
     "IS_T4_GPU",
     "GPU_NAME",

src/mosaic/telemetry/__init__.py

@@ -8,7 +8,7 @@ This module provides lightweight telemetry for the Mosaic HuggingFace app to tra
 Key features:
 - Gradio-only: No telemetry for CLI batch processing
 - No external dependencies: Uses stdlib only (json, dataclasses)
-- Privacy-first: Session IDs hashed, no file paths or PII stored
+- Privacy-first: Session IDs hashed, no file paths stored. HF usernames recorded for usage tracking.
 - File-based storage: JSONL format with daily rotation
 - HF Spaces compatible: Uses /data persistent storage when available
 

src/mosaic/telemetry/tracker.py

@@ -232,7 +232,7 @@ class TelemetryTracker:
             duration_sec: Analysis duration (for analysis_complete only)
             success: Whether analysis succeeded (for analysis_complete only)
             is_logged_in: True if HF user logged in
-            hf_username: Raw HF username (HF Spaces only)
+            hf_username: HF username (HF Spaces only)
         """
         if not self._is_enabled():
             return
@@ -298,7 +298,7 @@ class TelemetryTracker:
             gpu_type: GPU type string
             peak_gpu_memory_gb: Peak GPU memory usage in GB
             is_logged_in: True if HF user logged in
-            hf_username: Raw HF username (HF Spaces only)
+            hf_username: HF username (HF Spaces only)
         """
         if not self._is_enabled():
             return
@@ -349,7 +349,7 @@ class TelemetryTracker:
             slide_count: Number of slides if known
             gpu_type: GPU type string
             is_logged_in: True if HF user logged in
-            hf_username: Raw HF username (HF Spaces only)
+            hf_username: HF username (HF Spaces only)
         """
         if not self._is_enabled():
             return

src/mosaic/telemetry/utils.py

@@ -9,6 +9,7 @@ This module provides helper utilities:
 """
 
 import base64
+import binascii
 import hashlib
 import json
 import re
@@ -17,6 +18,8 @@ from contextlib import contextmanager
 from dataclasses import dataclass
 from typing import Dict, Optional
 
+from loguru import logger
+
 
 @contextmanager
 def StageTimer(stage_name: str, timings: Dict[str, float]):
@@ -130,7 +133,7 @@ def extract_user_info(request, is_hf_spaces: bool = False) -> UserInfo:
         UserInfo with is_logged_in and username (or defaults if extraction fails)
 
     Example:
-        user_info = extract_user_info(request, IS_ZEROGPU)
+        user_info = extract_user_info(request, IS_HF_SPACES)
         if user_info.is_logged_in:
             print(f"User: {user_info.username}")
     """
@@ -171,9 +174,9 @@ def extract_user_info(request, is_hf_spaces: bool = False) -> UserInfo:
             username = token_data["onBehalfOf"]["user"]
             return UserInfo(is_logged_in=True, username=username)
 
-    except Exception:
-        # Silently fail and return default UserInfo
-        # (Logging happens at call site if needed)
-        pass
+    except (json.JSONDecodeError, binascii.Error, UnicodeDecodeError, ValueError) as e:
+        logger.warning(f"Failed to decode JWT token from request: {e}")
+    except Exception as e:
+        logger.error(f"Unexpected error extracting user info: {e}")
 
     return UserInfo()

src/mosaic/ui/app.py

@@ -33,7 +33,7 @@ from mosaic.ui.utils import (
 )
 from mosaic.analysis import analyze_slide
 from mosaic.model_manager import load_all_models
-from mosaic.hardware import DEFAULT_CONCURRENCY_LIMIT, IS_T4_GPU, GPU_TYPE
+from mosaic.hardware import DEFAULT_CONCURRENCY_LIMIT, IS_HF_SPACES, IS_T4_GPU, GPU_TYPE
 from mosaic.telemetry import extract_user_info
 from mosaic.tcga import (
     fetch_slide,
@@ -218,7 +218,7 @@ def analyze_slides(
     session_hash = request.session_hash if request else None
 
     # Extract user info for telemetry (HF Spaces only)
-    user_info = extract_user_info(request, tracker.config.is_hf_spaces)
+    user_info = extract_user_info(request, IS_HF_SPACES)
 
     # Wait for core models download to complete (Paladin models can continue in background)
     if _model_download_thread is not None and not _core_models_complete:

tests/telemetry/test_events.py

@@ -108,6 +108,37 @@ class TestUsageEvent:
         assert data["slide_count"] == 3
         assert data["session_hash"] is None
 
+    def test_user_info_fields(self):
+        """Test is_logged_in and hf_username fields in UsageEvent."""
+        event = UsageEvent(
+            event_type="analysis_start",
+            analysis_id="test-123",
+            session_hash=None,
+            slide_count=1,
+            is_logged_in=True,
+            hf_username="testuser",
+        )
+
+        assert event.is_logged_in is True
+        assert event.hf_username == "testuser"
+
+        data = event.to_dict()
+        assert data["is_logged_in"] is True
+        assert data["hf_username"] == "testuser"
+
+    def test_user_info_fields_default_none(self):
+        """Test that user info fields default to None in to_dict()."""
+        event = UsageEvent(
+            event_type="analysis_start",
+            analysis_id="test-123",
+            session_hash=None,
+            slide_count=1,
+        )
+
+        data = event.to_dict()
+        assert data["is_logged_in"] is None
+        assert data["hf_username"] is None
+
 
 class TestResourceEvent:
     """Tests for ResourceEvent."""
@@ -147,6 +178,20 @@ class TestResourceEvent:
         assert data["total_duration_sec"] == 100.0
         assert data["tile_count"] == 500
 
+    def test_user_info_fields(self):
+        """Test is_logged_in and hf_username fields in ResourceEvent."""
+        event = ResourceEvent(
+            analysis_id="test-123",
+            session_hash=None,
+            total_duration_sec=100.0,
+            is_logged_in=True,
+            hf_username="testuser",
+        )
+
+        data = event.to_dict()
+        assert data["is_logged_in"] is True
+        assert data["hf_username"] == "testuser"
+
 
 class TestFailureEvent:
     """Tests for FailureEvent."""
@@ -190,6 +235,20 @@ class TestFailureEvent:
         assert data["error_type"] == "MemoryError"
         assert data["error_stage"] == "ctranspath"
 
+    def test_user_info_fields(self):
+        """Test is_logged_in and hf_username fields in FailureEvent."""
+        event = FailureEvent(
+            error_type="ValueError",
+            error_message="test error",
+            error_stage="upload",
+            is_logged_in=False,
+            hf_username=None,
+        )
+
+        data = event.to_dict()
+        assert data["is_logged_in"] is False
+        assert data["hf_username"] is None
+
 
 class TestEventIdGeneration:
     """Tests for event ID generation."""

tests/telemetry/test_tracker.py

@@ -137,6 +137,23 @@ class TestUsageEvents:
         assert event["session_hash"] is not None
         assert event["session_hash"] != "abc123"
 
+    def test_log_usage_event_with_user_info(self, tracker, temp_dir):
+        """Test that is_logged_in and hf_username are persisted in usage events."""
+        tracker.log_usage_event(
+            event_type="analysis_start",
+            analysis_id="test-user-info",
+            slide_count=1,
+            is_logged_in=True,
+            hf_username="testuser",
+        )
+
+        usage_files = list((temp_dir / "daily").glob("usage_*.jsonl"))
+        with open(usage_files[0]) as f:
+            event = json.loads(f.read().strip())
+
+        assert event["is_logged_in"] is True
+        assert event["hf_username"] == "testuser"
+
     def test_log_analysis_complete(self, tracker, temp_dir):
         """Test logging analysis complete event."""
         tracker.log_app_start()
@@ -206,6 +223,22 @@ class TestResourceEvents:
         assert event["tile_count"] == 1000
         assert event["peak_gpu_memory_gb"] == 12.5
 
+    def test_log_resource_event_with_user_info(self, tracker, temp_dir):
+        """Test that is_logged_in and hf_username are persisted in resource events."""
+        tracker.log_resource_event(
+            analysis_id="test-user-info",
+            total_duration_sec=60.0,
+            is_logged_in=True,
+            hf_username="testuser",
+        )
+
+        resource_files = list((temp_dir / "daily").glob("resource_*.jsonl"))
+        with open(resource_files[0]) as f:
+            event = json.loads(f.read().strip())
+
+        assert event["is_logged_in"] is True
+        assert event["hf_username"] == "testuser"
+
 
 class TestFailureEvents:
     """Tests for failure event logging."""
@@ -231,6 +264,23 @@ class TestFailureEvents:
         assert event["error_stage"] == "upload"
         assert event["analysis_id"] == "test-123"
 
+    def test_log_failure_event_with_user_info(self, tracker, temp_dir):
+        """Test that is_logged_in and hf_username are persisted in failure events."""
+        tracker.log_failure_event(
+            error_type="ValueError",
+            error_message="test error",
+            error_stage="upload",
+            is_logged_in=False,
+            hf_username=None,
+        )
+
+        failure_files = list((temp_dir / "daily").glob("failure_*.jsonl"))
+        with open(failure_files[0]) as f:
+            event = json.loads(f.read().strip())
+
+        assert event["is_logged_in"] is False
+        assert event["hf_username"] is None
+
     def test_error_message_sanitized(self, tracker, temp_dir):
         """Test that error messages are sanitized."""
         tracker.log_failure_event(

tests/telemetry/test_utils.py

@@ -302,3 +302,36 @@ class TestExtractUserInfo:
 
         assert user_info.is_logged_in is True
         assert user_info.username == "user-name_123"
+
+    def test_extract_user_info_jwt_on_behalf_of_without_user(self):
+        """Test extraction when onBehalfOf exists but user key is missing."""
+        header = {"alg": "HS256", "typ": "JWT"}
+        payload = {"onBehalfOf": {"role": "admin"}}  # No "user" key
+
+        header_b64 = base64.urlsafe_b64encode(json.dumps(header).encode()).decode()
+        payload_b64 = base64.urlsafe_b64encode(json.dumps(payload).encode()).decode()
+        header_b64 = header_b64.rstrip("=")
+        payload_b64 = payload_b64.rstrip("=")
+
+        token = f"{header_b64}.{payload_b64}.fake_sig"
+        referer = f"https://huggingface.co/spaces/test/app?__sign={token}"
+
+        request = self._create_mock_request(referer)
+        user_info = extract_user_info(request, is_hf_spaces=True)
+
+        assert user_info.is_logged_in is False
+        assert user_info.username is None
+
+    def test_extract_user_info_malformed_jwt_logs_warning(self, caplog):
+        """Test that malformed JWT triggers a warning log."""
+        import logging
+
+        referer = "https://huggingface.co/spaces/test/app?__sign=invalid.jwt.token"
+        request = self._create_mock_request(referer)
+
+        with caplog.at_level(logging.WARNING):
+            user_info = extract_user_info(request, is_hf_spaces=True)
+
+        assert user_info.is_logged_in is False
+        # loguru propagates to standard logging when caplog is used with propagate=True
+        # The function should log a warning about JWT decode failure