Record one summary per visit instead of every click; hourly writes to data/incoming; monthly compaction

DEPLOY.md

@@ -9,7 +9,7 @@ Prerequisites: `hf auth login` as a member of `mlx-community` with write access,
 ```bash
 .venv/bin/python -m pytest
 EXPLORER_SINK=local EXPLORER_FLUSH_SECONDS=20 .venv/bin/uvicorn app.main:app --port 7860
-# walk through the UI; events land in .runtime/local_dataset/data/events/
+# walk through the UI; events land in .runtime/local_dataset/data/incoming/
 docker build --platform linux/amd64 -t mlx-model-explorer . \
   && docker run --rm -p 7860:7860 -e EXPLORER_SINK=local mlx-model-explorer
 ```
@@ -21,7 +21,7 @@ docker build --platform linux/amd64 -t mlx-model-explorer . \
 .venv/bin/python scripts/deploy.py create      # refuses if the Space exists; add --allow-existing-dataset if the dataset was created earlier
 ```
 
-This creates `mlx-community/mlx-model-explorer-data` (private dataset, card uploaded) and `codelion/mlx-model-explorer` (private Docker Space with `EXPLORER_SINK=hub`, `DATASET_REPO`, `EXPLORER_FLUSH_SECONDS=600`).
+This creates `mlx-community/mlx-model-explorer-data` (private dataset, card uploaded) and `codelion/mlx-model-explorer` (private Docker Space with `EXPLORER_SINK=hub`, `DATASET_REPO`, `EXPLORER_FLUSH_SECONDS=3600`).
 
 > **Why the Space isn't in `mlx-community`:** Hugging Face now requires a Team or Enterprise plan for an organization to run Docker or Gradio Spaces on free CPU (`402 Payment Required`). The dataset can live in the org. Once an org admin has a plan or a hardware grant, deploy there with `SPACE_REPO=mlx-community/mlx-model-explorer scripts/deploy.py create --allow-existing-dataset` followed by `upload`.
 
@@ -82,6 +82,17 @@ hf upload mlx-community/mlx-model-explorer static_embed . --repo-type space
 
 Deployed as private `mlx-community/mlx-model-explorer` on 2026-09-14. The app renders inside the Hub page. Make the Static Space public at launch with `hf repos settings mlx-community/mlx-model-explorer --type space --no-private` (or from its Settings page).
 
+## Monthly compaction (run from any machine with write access)
+
+Early each month, merge the previous month's hourly shards into one file:
+
+```bash
+.venv/bin/python scripts/compact.py            # dry run: shows shards, rows before/after
+.venv/bin/python scripts/compact.py --apply    # one commit: add data/events/YYYY-MM.parquet, delete the month's data/incoming shards
+```
+
+It keeps the newest `session` row per visit, never edits row values, refuses the current month, and is safe to re-run (late shards are merged into the existing monthly file).
+
 ## Updating
 
 Change code, run the tests, `docker build` locally, then `scripts/deploy.py upload -m "what changed"`. Events are append-only, so redeploys never touch existing data. On shutdown the app flushes buffered events, and anything that fails to upload is retried from the on-disk spool.

README.md

@@ -46,14 +46,15 @@ app/         FastAPI
   memory.py         weights + KV cache + overhead, GPU-usable memory, fit classes
   recommend.py      RecommendationEngine interface + HeuristicEngine v1
   events.py         strict event schema, bounds, note scrubbing, plausibility flags
-  sink.py           append-only Parquet shards: local directory (dev) or Hub dataset (prod)
-  stats.py          k-anonymous aggregates + per-model community signals
+  sink.py           buffered Parquet shards into data/incoming: local directory (dev) or Hub dataset (prod)
+  stats.py          k-anonymous aggregates (one count per visit) + per-model community signals
 bench/       mlx_explorer_bench.py (runs locally with mlx-lm; submits only with --submit)
 data/        catalogue_snapshot.json.gz (fallback when the Hub API is unreachable)
+scripts/     deploy.py (create/upload/publish), compact.py (monthly compaction)
 tests/       pytest suite
 ```
 
-Events are buffered and written as **one Parquet shard per flush** (default every 10 minutes, only when there is data) under `data/events/YYYY/MM/DD/`. That caps Hub commits at about 144 a day. Shards are never rewritten, and a failed upload stays spooled on disk and retries with backoff.
+The page records **one summary row per visit** (final filters, what was searched, hardware class, models opened/compared/visited) plus explicit contributions (feedback, MLX and browser benchmarks). Individual clicks are never logged. The Space buffers rows and writes one Parquet shard to `data/incoming/YYYY/MM/DD/` at most hourly, or sooner after 2,000 rows. A failed upload stays spooled on disk and retries with backoff. Once a month, `scripts/compact.py` merges the previous month into `data/events/YYYY-MM.parquet` (see DEPLOY.md).
 
 ## Configuration
 
@@ -63,8 +64,9 @@ Events are buffered and written as **one Parquet shard per flush** (default ever
 | `DATASET_REPO` | `mlx-community/mlx-model-explorer-data` | dataset for `hub` mode |
 | `HF_TOKEN` | (none) | **Space secret**, a fine-grained token with write access to the dataset only. It is never used for catalogue requests |
 | `HF_READ_TOKEN` | (none) | optional token for Hub reads (rate limits only) |
-| `EXPLORER_FLUSH_SECONDS` | `600` | flush interval |
-| `EXPLORER_DATA_PREFIX` | `data/events` | path prefix inside the dataset (tests use `data/_test`) |
+| `EXPLORER_FLUSH_SECONDS` | `3600` | maximum time between writes |
+| `EXPLORER_FLUSH_ROWS` | `2000` | write early once this many rows are buffered |
+| `EXPLORER_DATA_PREFIX` | `data/incoming` | where the Space writes shards |
 | `EXPLORER_ORGS` | `mlx-community` | organizations to list |
 | `EXPLORER_RATE_PER_MIN` | `120` | event requests per client per minute |
 

app/events.py

@@ -1,9 +1,16 @@
 """Event schema, validation and server-side enrichment.
 
+The dataset records outcomes, not clicks:
+
+* ``session``: one row per visit (re-sent with the latest state; the newest row per
+  session_id wins), holding the final filters, what was searched, hardware class and
+  which models were opened, compared or visited.
+* ``feedback``, ``mlx_benchmark``, ``browser_benchmark``: explicit contributions.
+
 Clients send a small, typed event. Anything outside the schema is rejected
-(`extra="forbid"`), numbers are bounded, free text is capped and scrubbed, and the
-server fills in catalogue facts itself instead of trusting the client. Values that
-are allowed but implausible are kept and flagged, so dataset users can filter them.
+(``extra="forbid"``), numbers are bounded, free text is capped and scrubbed, and the
+server fills in catalogue facts itself instead of trusting the client. Values that are
+allowed but implausible are kept and flagged, so dataset users can filter them.
 """
 
 from __future__ import annotations
@@ -15,15 +22,16 @@ from typing import Literal
 
 from pydantic import BaseModel, ConfigDict, Field, field_validator
 
-from .memory import RAM_CLASSES, CONTEXTS, bits_per_weight
+from .memory import CONTEXTS, RAM_CLASSES, bits_per_weight
 from .parsing import PARAM_BUCKETS, QUANT_BUCKETS
 
-SCHEMA_VERSION = 1
+SCHEMA_VERSION = 2
 
-EventType = Literal[
-    "search", "filter", "model_view", "model_select", "model_click", "compare",
-    "hardware_test", "browser_benchmark", "feedback", "mlx_benchmark_submission",
-]
+EventType = Literal["session", "feedback", "mlx_benchmark", "browser_benchmark"]
+# Accepted from older clients and the published benchmark script, stored under the new name.
+ALIASES = {"mlx_benchmark_submission": "mlx_benchmark"}
+# Clickstream events from pages cached before the redesign: accepted and dropped.
+LEGACY_DROPPED = {"search", "filter", "model_view", "model_select", "model_click", "compare", "hardware_test"}
 
 CHIP_RE = re.compile(r"^Apple M[1-9]( (Pro|Max|Ultra))?$")
 MODEL_ID_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._-]{0,95}/[A-Za-z0-9][A-Za-z0-9._-]{0,95}$")
@@ -36,7 +44,8 @@ _URL = re.compile(r"(https?://|www\.)\S+", re.I)
 _LONG_DIGITS = re.compile(r"\+?\d[\d ()-]{7,}\d")
 
 MAX_NOTES = 280
-MAX_BATCH = 50
+MAX_BATCH = 20
+QuantBucket = Literal[tuple(QUANT_BUCKETS)]  # type: ignore[valid-type]
 
 
 def clean_notes(text: str | None) -> str | None:
@@ -51,21 +60,37 @@ def clean_notes(text: str | None) -> str | None:
     return text[:MAX_NOTES] or None
 
 
+def _model_ids(v, cap):
+    if v is None:
+        return v
+    if len(v) > cap:
+        raise ValueError(f"at most {cap} model ids")
+    for m in v:
+        if not MODEL_ID_RE.match(m):
+            raise ValueError("invalid model id")
+    return list(dict.fromkeys(v))  # de-duplicate, keep order
+
+
 class ClientEvent(BaseModel):
     """What the browser (or the benchmark script) is allowed to send."""
 
     model_config = ConfigDict(extra="forbid", str_max_length=200)
 
-    event_type: EventType
+    event_type: str
     session_id: str | None = None
 
+    # the visit's final query
     model_family: str | None = Field(None, max_length=40)
     parameter_bucket: Literal[tuple(PARAM_BUCKETS + ["MoE", "unknown"])] | None = None  # type: ignore[valid-type]
-    quantization: Literal[tuple(QUANT_BUCKETS)] | None = None  # type: ignore[valid-type]
+    quantization: QuantBucket | None = None
     target_context: int | None = None
     priority: Literal["balanced", "quality", "speed", "memory", "long_context"] | None = None
     sort: Literal["recommended", "popular", "recent", "community", "all"] | None = None
     result_count: int | None = Field(None, ge=0, le=100_000)
+    # everything the visit looked at along the way
+    families_searched: list[str] | None = Field(None, max_length=5)
+    quants_searched: list[QuantBucket] | None = Field(None, max_length=5)
+    distinct_queries: int | None = Field(None, ge=0, le=10_000)
 
     hardware_source: Literal["detected", "confirmed", "none"] | None = None
     hardware_memory_class: int | None = None
@@ -78,13 +103,16 @@ class ClientEvent(BaseModel):
     os_family: Literal["macos", "ios", "windows", "linux", "android", "other"] | None = None
     cpu_cores: int | None = Field(None, ge=1, le=256)
 
-    selected_model: str | None = None
-    selected_model_rank: int | None = Field(None, ge=0, le=10_000)
-    # what the user was shown for that model at the time (for studying recommendations)
-    recommendation_score: float | None = Field(None, ge=0, le=100)
-    fit_class: Literal["Comfortable", "Likely", "Borderline", "Unlikely"] | None = None
-    compare_models: list[str] | None = Field(None, max_length=5)
+    # what the visit engaged with
+    top_model: str | None = None
+    top_model_score: float | None = Field(None, ge=0, le=100)
+    top_model_fit: Literal["Comfortable", "Likely", "Borderline", "Unlikely"] | None = None
+    models_viewed: list[str] | None = None
+    models_compared: list[str] | None = None
+    models_clicked: list[str] | None = None
 
+    # feedback and benchmarks are about one model
+    selected_model: str | None = None
     tried: Literal["yes", "no", "planning"] | None = None
     quality_rating: Literal["excellent", "good", "acceptable"] | None = None
     failure_reason: Literal["too_slow", "too_much_memory", "low_quality", "didnt_run"] | None = None
@@ -112,6 +140,14 @@ class ClientEvent(BaseModel):
     mlx_lm_version: str | None = None
     macos_major: int | None = Field(None, ge=11, le=40)
 
+    @field_validator("event_type")
+    @classmethod
+    def _type(cls, v):
+        v = ALIASES.get(v, v)
+        if v not in ("session", "feedback", "mlx_benchmark", "browser_benchmark") and v not in LEGACY_DROPPED:
+            raise ValueError("unknown event_type")
+        return v
+
     @field_validator("session_id")
     @classmethod
     def _session(cls, v):
@@ -119,22 +155,38 @@ class ClientEvent(BaseModel):
             raise ValueError("session_id must be 16-32 lowercase hex chars")
         return v
 
-    @field_validator("selected_model")
+    @field_validator("selected_model", "top_model")
     @classmethod
     def _model(cls, v):
         if v is not None and not MODEL_ID_RE.match(v):
             raise ValueError("invalid model id")
         return v
 
-    @field_validator("compare_models")
+    @field_validator("models_viewed")
     @classmethod
-    def _models(cls, v):
+    def _viewed(cls, v):
+        return _model_ids(v, 10)
+
+    @field_validator("models_compared", "models_clicked")
+    @classmethod
+    def _compared(cls, v):
+        return _model_ids(v, 5)
+
+    @field_validator("families_searched")
+    @classmethod
+    def _families(cls, v):
         if v is not None:
-            for m in v:
-                if not MODEL_ID_RE.match(m):
-                    raise ValueError("invalid model id in compare_models")
+            for f in v:
+                if not SHORT_TOKEN_RE.match(f):
+                    raise ValueError("invalid family")
+            v = list(dict.fromkeys(v))
         return v
 
+    @field_validator("quants_searched")
+    @classmethod
+    def _quants(cls, v):
+        return list(dict.fromkeys(v)) if v is not None else v
+
     @field_validator("target_context")
     @classmethod
     def _ctx(cls, v):
@@ -183,28 +235,34 @@ class EventBatch(BaseModel):
 
 # Column order for the dataset. Every row has every column (null when unused).
 COLUMNS: dict[str, str] = {
+    # every row
     "timestamp": "string", "schema_version": "int", "app_version": "string", "event_type": "string",
     "session_id": "string",
-    "model_family": "string", "model_name": "string", "parameter_bucket": "string", "quantization": "string",
-    "quant_bits": "float", "quant_mixed": "bool",
+    # session: final query and what was searched
+    "model_family": "string", "parameter_bucket": "string", "quantization": "string",
     "target_context": "int", "priority": "string", "sort": "string", "result_count": "int",
+    "families_searched": "list", "quants_searched": "list", "distinct_queries": "int",
+    # session + benchmarks: hardware class
     "hardware_source": "string", "hardware_memory_class": "int", "hardware_confirmed": "bool",
     "webgpu_available": "bool", "webgpu_score": "float", "gpu_capability_class": "string",
     "gpu_vendor": "string", "gpu_arch": "string", "browser_family": "string", "os_family": "string",
     "cpu_cores": "int",
-    "selected_model": "string", "selected_model_rank": "int", "compare_models": "list",
-    "hf_downloads_at_selection": "int", "hf_likes_at_selection": "int",
-    "recommendation_score": "float", "fit_class": "string", "engine_version": "string",
-    "pipeline": "string",
+    # session: engagement
+    "top_model": "string", "top_model_score": "float", "top_model_fit": "string", "engine_version": "string",
+    "models_viewed": "list", "models_compared": "list", "models_clicked": "list",
+    # feedback / mlx_benchmark: the model, with catalogue facts filled in by the server
+    "selected_model": "string", "model_name": "string", "quant_bits": "float", "quant_mixed": "bool",
+    "pipeline": "string", "hf_downloads_at_selection": "int", "hf_likes_at_selection": "int",
+    # feedback
     "tried": "string", "outcome": "string", "quality_rating": "string", "failure_reason": "string",
     "reported_ram_gb": "int", "reported_mac_model": "string", "reported_tokens_per_second": "float",
     "reported_context": "int", "notes": "string",
+    # benchmarks
     "benchmark_type": "string", "benchmark_version": "string", "benchmark_duration_ms": "int",
     "prompt_tokens": "int", "generation_tokens": "int", "prompt_tps": "float", "generation_tps": "float",
     "ttft_ms": "float", "peak_memory_gb": "float", "chip": "string",
     "perplexity": "float", "perplexity_stderr": "float", "eval_dataset": "string", "eval_tokens": "int",
-    "mlx_version": "string",
-    "mlx_lm_version": "string", "macos_major": "int",
+    "mlx_version": "string", "mlx_lm_version": "string", "macos_major": "int",
     "suspicious_flags": "list",
 }
 
@@ -213,8 +271,10 @@ def utc_now_seconds() -> str:
     return datetime.now(timezone.utc).replace(microsecond=0).isoformat().replace("+00:00", "Z")
 
 
-def to_row(ev: ClientEvent, catalogue=None, app_version: str = "dev", engine_version: str | None = None) -> dict:
-    """Validated client event -> dataset row, with server-side facts and suspicion flags."""
+def to_row(ev: ClientEvent, catalogue=None, app_version: str = "dev", engine_version: str | None = None) -> dict | None:
+    """Validated client event -> dataset row (None for legacy clickstream events, which are dropped)."""
+    if ev.event_type in LEGACY_DROPPED:
+        return None
     d = ev.model_dump()
     row = {k: None for k in COLUMNS}
     for k, v in d.items():
@@ -223,8 +283,9 @@ def to_row(ev: ClientEvent, catalogue=None, app_version: str = "dev", engine_ver
     row["timestamp"] = utc_now_seconds()
     row["schema_version"] = SCHEMA_VERSION
     row["app_version"] = app_version
-    row["engine_version"] = engine_version
     row["hardware_confirmed"] = ev.hardware_source == "confirmed" if ev.hardware_source else None
+    if ev.event_type == "session":
+        row["engine_version"] = engine_version
     if ev.tried == "yes":
         if ev.failure_reason:
             row["outcome"] = "problem"
@@ -247,9 +308,11 @@ def to_row(ev: ClientEvent, catalogue=None, app_version: str = "dev", engine_ver
             row["hf_downloads_at_selection"] = rec.downloads
             row["hf_likes_at_selection"] = rec.likes
             row["pipeline"] = rec.pipeline
-    if ev.compare_models and catalogue is not None:
-        if any(m not in catalogue.by_id for m in ev.compare_models):
-            flags.append("unknown_model_in_compare")
+    if catalogue is not None and ev.event_type == "session":
+        listed = [m for m in [ev.top_model, *(ev.models_viewed or []), *(ev.models_compared or []),
+                              *(ev.models_clicked or [])] if m]
+        if any(m not in catalogue.by_id for m in listed):
+            flags.append("unknown_model_in_session")
 
     flags += plausibility_flags(ev, rec)
     row["suspicious_flags"] = sorted(set(flags))
@@ -276,13 +339,15 @@ def plausibility_flags(ev: ClientEvent, rec=None) -> list[str]:
     ram = ev.reported_ram_gb or ev.hardware_memory_class
     if ev.peak_memory_gb and ram and ev.peak_memory_gb > ram * 1.05:
         flags.append("peak_memory_exceeds_ram")
-    if ev.event_type == "mlx_benchmark_submission":
+    if ev.event_type == "mlx_benchmark":
         if not (ev.selected_model and (ev.generation_tps or ev.perplexity) and ev.benchmark_version):
             flags.append("incomplete_benchmark")
         if ev.benchmark_type not in (None, "mlx_lm"):
             flags.append("benchmark_type_mismatch")
     if ev.event_type == "browser_benchmark" and ev.benchmark_type == "mlx_lm":
         flags.append("benchmark_type_mismatch")
+    if ev.event_type == "feedback" and not ev.selected_model:
+        flags.append("incomplete_feedback")
     if ev.perplexity is not None and not (ev.eval_dataset and ev.eval_tokens and ev.selected_model):
         flags.append("incomplete_quality")
     if ev.perplexity is not None and ev.perplexity > 1000:
@@ -292,3 +357,22 @@ def plausibility_flags(ev: ClientEvent, rec=None) -> list[str]:
     if ev.failure_reason and ev.quality_rating:
         flags.append("conflicting_feedback")
     return flags
+
+
+def latest_sessions(rows: list[dict]) -> list[dict]:
+    """Collapse re-sent session rows to the newest per session_id; other rows pass through.
+
+    Timestamps have one-second precision, so rows later in the input win ties (the
+    sink and the compactor both preserve arrival order).
+    """
+    out, latest = [], {}
+    for i, r in enumerate(rows):
+        if r.get("event_type") == "session" and r.get("session_id"):
+            key = r["session_id"]
+            prev = latest.get(key)
+            if prev is None or (r.get("timestamp") or "") >= (rows[prev].get("timestamp") or ""):
+                latest[key] = i
+        else:
+            out.append(r)
+    out.extend(rows[i] for i in sorted(latest.values()))
+    return sorted(out, key=lambda r: r.get("timestamp") or "")

app/main.py

@@ -270,9 +270,11 @@ def create_app(state: State | None = None) -> FastAPI:
         if not st.collection_enabled:
             return {"accepted": 0}
         rows = [to_row(ev, st.catalogue, APP_VERSION, st.engine.version) for ev in batch.events]
-        st.sink.add(rows)
-        st.stats.invalidate()
-        return {"accepted": len(rows), "flags": [r["suspicious_flags"] for r in rows]}
+        kept = [r for r in rows if r is not None]  # legacy clickstream events from cached pages are dropped
+        if kept:
+            st.sink.add(kept)
+            st.stats.invalidate()
+        return {"accepted": len(kept), "flags": [r["suspicious_flags"] if r else ["dropped_legacy_event"] for r in rows]}
 
     @app.get("/api/stats")
     def stats():

app/sink.py

@@ -1,9 +1,11 @@
-"""Append-only event storage.
-
-Rows are buffered in memory and mirrored to a local spool file. On each flush the
-buffer becomes ONE new Parquet shard (one Hub commit), so a busy day costs at most
-~144 commits at the default 10-minute interval. Existing shards are never
-rewritten. If an upload fails, the rows stay spooled and go out with the next flush.
+"""Event storage.
+
+The Space buffers rows in memory, mirrored to a local spool file, and writes them as ONE
+new Parquet shard under ``data/incoming/`` per flush: hourly, or sooner once the buffer
+reaches ``FLUSH_ROWS``. That is at most ~24 small files a day. Once a month
+``scripts/compact.py`` merges a finished month into ``data/events/YYYY-MM.parquet`` and
+removes its incoming shards in the same commit. If an upload fails, the rows stay spooled
+and go out with the next flush.
 """
 
 from __future__ import annotations
@@ -29,6 +31,9 @@ _PA = {"string": pa.string(), "int": pa.int64(), "float": pa.float64(), "bool":
        "list": pa.list_(pa.string())}
 SCHEMA = pa.schema([(k, _PA[t]) for k, t in COLUMNS.items()])
 MAX_BUFFER = 100_000
+FLUSH_ROWS = int(os.environ.get("EXPLORER_FLUSH_ROWS", "2000"))
+INCOMING = "data/incoming"
+MONTHLY = "data/events"
 
 
 def rows_to_parquet_bytes(rows: list[dict]) -> bytes:
@@ -47,7 +52,7 @@ def shard_path(prefix: str, instance: str, when: datetime | None = None) -> str:
 class EventSink:
     kind = "base"
 
-    def __init__(self, spool_dir: Path, flush_seconds: float = 600, prefix: str = "data/events"):
+    def __init__(self, spool_dir: Path, flush_seconds: float = 3600, prefix: str = INCOMING):
         self.prefix = prefix
         self.flush_seconds = flush_seconds
         self.instance = uuid.uuid4().hex[:8]
@@ -94,6 +99,9 @@ class EventSink:
                         f.write(json.dumps(r) + "\n")
             except OSError:
                 log.warning("spool write failed", exc_info=True)
+            big = len(self._buf) >= FLUSH_ROWS
+        if big and not self._flush_lock.locked():
+            threading.Thread(target=self.flush, daemon=True, name="sink-size-flush").start()
 
     @property
     def pending(self) -> int:
@@ -183,9 +191,10 @@ class LocalParquetSink(EventSink):
 
     def read_existing(self) -> list[dict]:
         rows = []
-        for p in sorted((self.root / self.prefix).rglob("*.parquet")):
+        paths = sorted((self.root / MONTHLY).glob("*.parquet")) + sorted((self.root / self.prefix).rglob("*.parquet"))
+        for p in dict.fromkeys(paths):
             try:
-                rows.extend(pq.read_table(p).to_pylist())
+                rows.extend(_aligned(pq.read_table(p).to_pylist()))
             except Exception:
                 log.warning("unreadable shard %s", p)
         return rows
@@ -216,15 +225,18 @@ class HubParquetSink(EventSink):
 
         rows = []
         try:
-            files = [f for f in self.api.list_repo_files(self.repo_id, repo_type="dataset")
-                     if f.startswith(self.prefix.strip("/") + "/") and f.endswith(".parquet")]
+            all_files = self.api.list_repo_files(self.repo_id, repo_type="dataset")
+            monthly = sorted(f for f in all_files if f.startswith(MONTHLY + "/") and f.count("/") == 2
+                             and f.endswith(".parquet"))
+            incoming = sorted(f for f in all_files if f.startswith(self.prefix.strip("/") + "/") and f.endswith(".parquet"))
+            files = list(dict.fromkeys(monthly + incoming))
         except Exception as e:
             self.last_error = f"read: {type(e).__name__}"
             return rows
         for f in files:
             try:
                 local = hf_hub_download(self.repo_id, f, repo_type="dataset", token=self.api.token)
-                rows.extend(pq.read_table(local).to_pylist())
+                rows.extend(_aligned(pq.read_table(local).to_pylist()))
             except Exception:
                 log.warning("unreadable shard %s", f)
         return rows
@@ -242,10 +254,15 @@ class NullSink(EventSink):
         pass
 
 
+def _aligned(rows: list[dict]) -> list[dict]:
+    """Rows in the current column set (older shards may lack newer columns)."""
+    return [{k: r.get(k) for k in COLUMNS} for r in rows]
+
+
 def make_sink(base_dir: Path) -> EventSink:
     mode = os.environ.get("EXPLORER_SINK", "local")
-    flush = float(os.environ.get("EXPLORER_FLUSH_SECONDS", "600"))
-    prefix = os.environ.get("EXPLORER_DATA_PREFIX", "data/events")
+    flush = float(os.environ.get("EXPLORER_FLUSH_SECONDS", "3600"))
+    prefix = os.environ.get("EXPLORER_DATA_PREFIX", INCOMING)
     if mode == "hub":
         repo = os.environ.get("DATASET_REPO", "mlx-community/mlx-model-explorer-data")
         return HubParquetSink(repo, os.environ.get("HF_TOKEN"), spool_dir=base_dir / "_spool",

app/stats.py

@@ -1,6 +1,7 @@
 """Aggregate statistics. Only counts and shares leave this module, never rows.
 
-Any bucket seen fewer than K_MIN times is folded into "other", so a rare
+Sessions are counted once (the newest row per session_id), flagged rows are left out,
+and any bucket seen fewer than K_MIN times is folded into "other", so a rare
 configuration can't be traced back to one visitor.
 """
 
@@ -12,9 +13,11 @@ import time
 from collections import Counter, defaultdict
 from typing import Callable, Iterable
 
+from .events import latest_sessions
 from .recommend import EVAL_DATASET, CommunitySignal
 
 K_MIN = 5
+BENCH_TYPES = ("mlx_benchmark", "mlx_benchmark_submission")
 
 
 def _suppress(counter: Counter, k: int = K_MIN, top: int = 15) -> list[dict]:
@@ -30,62 +33,46 @@ def _suppress(counter: Counter, k: int = K_MIN, top: int = 15) -> list[dict]:
     return items
 
 
+def _present(sessions: list[dict], key: str) -> Counter:
+    """Shares among visits that set the field (unset isn't a category)."""
+    return Counter(s[key] for s in sessions if s.get(key) is not None)
+
+
 def compute(rows: Iterable[dict], k: int = K_MIN) -> dict:
-    rows = [r for r in rows if not r.get("suspicious_flags")]
+    rows = latest_sessions([r for r in rows if not r.get("suspicious_flags")])
+    sessions = [r for r in rows if r.get("event_type") == "session"]
     by_type = Counter(r.get("event_type") for r in rows)
-    sessions = len({r.get("session_id") for r in rows if r.get("session_id")})
-    # A "query" is a distinct configuration a session looked at (explicit search, or re-running
-    # it after changing sort/hardware). Counting each once keeps re-renders from inflating shares.
-    seen, searches = set(), []
-    for r in rows:
-        if r.get("event_type") in ("search", "filter"):
-            key = (r.get("session_id") or id(r), r.get("model_family"), r.get("parameter_bucket"),
-                   r.get("quantization"), r.get("target_context"), r.get("priority"), r.get("hardware_memory_class"))
-            if key not in seen:
-                seen.add(key)
-                searches.append(r)
-
-    quant = Counter(r.get("quantization") for r in searches if r.get("quantization"))
-    qtotal = sum(quant.values())
-    share = lambda b: round(quant.get(b, 0) / qtotal, 4) if qtotal else None
-    contexts = [r["target_context"] for r in searches if r.get("target_context")]
-
-    hw_seen, hw = set(), []
-    for r in rows:  # one count per session and memory class
-        if r.get("hardware_memory_class"):
-            key = (r.get("session_id") or id(r), r["hardware_memory_class"])
-            if key not in hw_seen:
-                hw_seen.add(key)
-                hw.append(r)
+
+    quants = Counter(q for s in sessions for q in (s.get("quants_searched") or []))
+    qtotal = sum(v for key, v in quants.items() if key != "unknown")
+    share = lambda b: round(quants.get(b, 0) / qtotal, 4) if qtotal else None
+    contexts = [s["target_context"] for s in sessions if s.get("target_context")]
+    engaged = [s for s in sessions if s.get("models_viewed") or s.get("models_compared") or s.get("models_clicked")]
+    picked = Counter(m for s in sessions for m in dict.fromkeys(
+        [*(s.get("models_compared") or []), *(s.get("models_clicked") or [])]))
+    hw = [s for s in sessions if s.get("gpu_capability_class")]
+
     return {
         "k_min": k,
         "totals": {
-            "events": len(rows),
-            "sessions": sessions,
-            "searches": by_type.get("search", 0),
-            "distinct_queries": len(searches),
-            "model_views": by_type.get("model_view", 0),
-            "model_clicks": by_type.get("model_click", 0),
-            "hardware_tests": by_type.get("hardware_test", 0),
+            "sessions": len(sessions),
+            "engaged_sessions": len(engaged),
+            "hardware_checks": len(hw),
             "browser_benchmarks": by_type.get("browser_benchmark", 0),
             "feedback": by_type.get("feedback", 0),
-            "mlx_benchmarks": by_type.get("mlx_benchmark_submission", 0),
+            "mlx_benchmarks": sum(by_type.get(t, 0) for t in BENCH_TYPES),
         },
-        "quant_share": {"2-bit": share("2-bit"), "3-bit": share("3-bit"), "4-bit": share("4-bit"),
-                        "5-bit": share("5-bit"), "6-bit": share("6-bit"), "8-bit": share("8-bit")},
+        "quant_share": {b: share(b) for b in ("2-bit", "3-bit", "4-bit", "5-bit", "6-bit", "8-bit")},
         "average_target_context": round(statistics.mean(contexts)) if len(contexts) >= k else None,
-        "families": _suppress(Counter(r.get("model_family") for r in searches), k),
-        "quantizations": _suppress(quant, k),
-        "sizes": _suppress(Counter(r.get("parameter_bucket") for r in searches), k),
-        "ram_classes": _suppress(Counter(r.get("hardware_memory_class") for r in hw), k),
-        "contexts": _suppress(Counter(r.get("target_context") for r in searches), k),
-        "priorities": _suppress(Counter(r.get("priority") for r in searches), k),
-        "selected_models": _suppress(Counter(r.get("selected_model") for r in rows
-                                             if r.get("event_type") in ("model_select", "model_click")), k),
-        "gpu_capability": _suppress(Counter(r.get("gpu_capability_class") for r in rows
-                                            if r.get("event_type") == "hardware_test"), k),
-        "webgpu_available": _suppress(Counter(r.get("webgpu_available") for r in rows
-                                              if r.get("event_type") == "hardware_test"), k),
+        "families": _suppress(Counter(f for s in sessions for f in (s.get("families_searched") or [])), k),
+        "quantizations": _suppress(quants, k),
+        "sizes": _suppress(_present(sessions, "parameter_bucket"), k),
+        "ram_classes": _suppress(_present(sessions, "hardware_memory_class"), k),
+        "contexts": _suppress(_present(sessions, "target_context"), k),
+        "priorities": _suppress(_present(sessions, "priority"), k),
+        "selected_models": _suppress(picked, k),
+        "gpu_capability": _suppress(Counter(s.get("gpu_capability_class") for s in hw), k),
+        "webgpu_available": _suppress(Counter(s.get("webgpu_available") for s in hw), k),
     }
 
 
@@ -101,7 +88,7 @@ def community_signals(rows: Iterable[dict]) -> dict[str, CommunitySignal]:
             continue
         if r.get("event_type") == "feedback" and r.get("tried") == "yes":
             fb[m].append(r)
-        elif r.get("event_type") == "mlx_benchmark_submission":
+        elif r.get("event_type") in BENCH_TYPES:
             if r.get("generation_tps"):
                 bench[m].append(float(r["generation_tps"]))
             if r.get("perplexity") and r.get("eval_dataset") == EVAL_DATASET:

bench/mlx_explorer_bench.py

@@ -155,7 +155,7 @@ def run(args) -> dict:
     med = lambda xs: statistics.median(xs)
     mac = platform.mac_ver()[0]
     return result | {
-        "event_type": "mlx_benchmark_submission",
+        "event_type": "mlx_benchmark",
         "benchmark_type": "mlx_lm",
         "benchmark_version": BENCHMARK_VERSION,
         "selected_model": args.model,

static/app.js

@@ -76,23 +76,35 @@
       sid = [...b].map((x) => x.toString(16).padStart(2, "0")).join("");
       try { sessionStorage.setItem("mme-sid", sid); } catch (e) {}
     }
-    let queue = [];
     let enabled = true;
+    let buildSession = () => null;  // set by the app: returns the visit's current summary or null
+    let lastSent = "";
 
-    function send(beacon) {
-      if (!queue.length || optedOut || !enabled) { queue = []; return; }
-      const batch = queue.splice(0, 50);
-      const body = JSON.stringify({ events: batch });
+    function post(events, beacon) {
+      const body = JSON.stringify({ events });
       if (beacon && navigator.sendBeacon) {
         navigator.sendBeacon("/api/events", new Blob([body], { type: "application/json" }));
       } else {
         fetch("/api/events", { method: "POST", headers: { "content-type": "application/json" }, body, keepalive: true }).catch(() => {});
       }
-      if (queue.length) send(beacon);
     }
-    setInterval(() => send(false), 10000);
-    document.addEventListener("visibilitychange", () => { if (document.visibilityState === "hidden") send(true); });
-    window.addEventListener("pagehide", () => send(true));
+
+    // One row per visit: re-sent only when the summary changed since the last send.
+    function sendSession(beacon) {
+      if (optedOut || !enabled) return;
+      const summary = buildSession();
+      if (!summary) return;
+      const ev = { event_type: "session", session_id: sid };
+      for (const [k, v] of Object.entries(summary)) {
+        if (v !== undefined && v !== null && v !== "" && !(Array.isArray(v) && !v.length)) ev[k] = v;
+      }
+      const key = JSON.stringify(ev);
+      if (key === lastSent) return;
+      lastSent = key;
+      post([ev], beacon);
+    }
+    document.addEventListener("visibilitychange", () => { if (document.visibilityState === "hidden") sendSession(true); });
+    window.addEventListener("pagehide", () => sendSession(true));
 
     return {
       gpc,
@@ -103,15 +115,16 @@
           if (v) { localStorage.setItem("mme-optout", "1"); localStorage.removeItem("mme-optin"); }
           else { localStorage.removeItem("mme-optout"); if (gpc) localStorage.setItem("mme-optin", "1"); }
         } catch (e) {}
-        if (optedOut) queue = [];
       },
       setEnabled(v) { enabled = v; },
-      track(type, fields) {
+      setSessionBuilder(fn) { buildSession = fn; },
+      sendSession,
+      // A contribution the page makes on the user's behalf (the GPU test): one row, sent now.
+      contribute(type, fields) {
         if (optedOut || !enabled) return;
         const ev = { event_type: type, session_id: sid };
         for (const [k, v] of Object.entries(fields || {})) if (v !== undefined && v !== null && v !== "") ev[k] = v;
-        queue.push(ev);
-        if (queue.length >= 20) send(false);
+        post([ev], false);
       },
       // Submissions the user explicitly asked to send go out immediately, with a result.
       async submit(type, fields) {
@@ -154,6 +167,43 @@
     $$(`#query input[name="${name}"]`).forEach((x) => (x.checked = false));
   }
 
+  // ------------------------------------------------------------------ visit summary (the only analytics row)
+  const visit = { families: [], quants: [], queries: new Set(), query: null, top: null,
+                  viewed: [], compared: [], clicked: [], engagedSent: false };
+  const pushCapped = (arr, v, cap) => { if (v && !arr.includes(v)) { arr.push(v); if (arr.length > cap) arr.shift(); } };
+
+  function noteQuery(q, data) {
+    visit.query = { ...q, result_count: data.total };
+    if (q.family) pushCapped(visit.families, q.family, 5);
+    if (q.quant) pushCapped(visit.quants, q.quant, 5);
+    visit.queries.add([q.family, q.size, q.quant, q.context, q.priority, q.ram_gb].join("|"));
+    const top = data.results && data.results[0];
+    if (top && q.sort === "recommended") {
+      visit.top = { id: top.model.id, score: Math.round(top.score * 10) / 10, fit: top.memory && top.memory.fit };
+    }
+  }
+
+  function noteModel(kind, id) {
+    pushCapped(visit[kind], id, kind === "viewed" ? 10 : 5);
+    if (!visit.engagedSent) {  // the first real interaction sends the summary right away
+      visit.engagedSent = true;
+      Telemetry.sendSession(false);
+    }
+  }
+
+  Telemetry.setSessionBuilder(() => {
+    const q = visit.query;
+    if (!q) return null;
+    return {
+      model_family: q.family, parameter_bucket: q.size, quantization: q.quant, target_context: q.context,
+      priority: q.priority, sort: q.sort, result_count: q.result_count,
+      families_searched: visit.families, quants_searched: visit.quants, distinct_queries: visit.queries.size,
+      ...hwFields(), webgpu_score: state.hw ? state.hw.quick_score : null,
+      top_model: visit.top && visit.top.id, top_model_score: visit.top && visit.top.score, top_model_fit: visit.top && visit.top.fit,
+      models_viewed: visit.viewed, models_compared: visit.compared, models_clicked: visit.clicked,
+    };
+  });
+
   function currentRam() {
     const v = parseInt(radioValue("ram"), 10);
     return Number.isFinite(v) ? v : null;
@@ -350,12 +400,7 @@
       state.results = append ? state.results.concat(data.results) : data.results;
       renderResults(data, append);
       syncUrl(q);
-      if (!append) {
-        Telemetry.track(reason, {
-          model_family: q.family, parameter_bucket: q.size, quantization: q.quant, target_context: q.context,
-          priority: q.priority, sort: q.sort, result_count: data.total, ...hwFields(),
-        });
-      }
+      if (!append) noteQuery(q, data);
     } catch (e) {
       if (id !== state.requestId) return;
       $("#summary").textContent = "";
@@ -439,13 +484,8 @@
     return r.reasons.find((t) => !skip.test(t)) || r.reasons[0] || "";
   }
 
-  function shown(r) {
-    return r ? { recommendation_score: Math.round(r.score * 10) / 10, fit_class: r.memory && r.memory.fit } : {};
-  }
-
-  function trackClick(id, rank, r) {
-    const q = state.lastQuery || readQuery();
-    Telemetry.track("model_click", { selected_model: id, selected_model_rank: rank, target_context: q.context, priority: q.priority, ...shown(r), ...hwFields() });
+  function trackClick(id) {
+    noteModel("clicked", id);
   }
 
   // ------------------------------------------------------------------ compare
@@ -457,8 +497,7 @@
         return;
       }
       state.compare.set(r.model.id, r);
-      const q = state.lastQuery || readQuery();
-      Telemetry.track("model_select", { selected_model: r.model.id, selected_model_rank: rank, target_context: q.context, priority: q.priority, ...shown(r), ...hwFields() });
+      noteModel("compared", r.model.id);
     } else {
       state.compare.delete(r.model.id);
     }
@@ -492,7 +531,6 @@
   function openCompare() {
     const items = [...state.compare.values()];
     const q = state.lastQuery || readQuery();
-    Telemetry.track("compare", { compare_models: items.map((x) => x.model.id), target_context: q.context, priority: q.priority, ...hwFields() });
     const scale = scaleFor(q.ram_gb, items.map((r) => r.memory.total_gb));
     const qt = (r) => r.quant || r.model.quant;
     const rows = [
@@ -527,7 +565,7 @@
     $("#d-title").textContent = id.split("/")[1];
     $("#d-body").replaceChildren(el("p", { class: "fine" }, "Loading model details"));
     if (!dlg.open) dlg.showModal();
-    Telemetry.track("model_view", { selected_model: id, selected_model_rank: rank ?? null, target_context: q.context, priority: q.priority, ...shown(r), ...hwFields() });
+    noteModel("viewed", id);
     const params = new URLSearchParams({ context: q.context, priority: q.priority });
     if (q.ram_gb) { params.set("ram_gb", q.ram_gb); if (q.ram_source) params.set("ram_source", q.ram_source); }
     let d;
@@ -620,6 +658,7 @@
       const tried = ($("input[name=tried]:checked", f) || {}).value;
       if (!tried) { out.textContent = "Choose Yes, No or Planning to."; return; }
       const fields = { selected_model: m.id, tried, target_context: q.context, priority: q.priority, ...hwFields() };
+      delete fields.cpu_cores;
       if (tried === "yes") {
         const oc = ($("input[name=outcome]:checked", f) || {}).value;
         if (oc && oc.startsWith("q:")) fields.quality_rating = oc.slice(2);
@@ -678,12 +717,6 @@
     }
     if (msgs.length) { $("#hw-error").textContent = msgs.join(" "); $("#hw-error").hidden = false; }
     $("#full-bench").disabled = !hw.webgpu_available;
-    Telemetry.track("hardware_test", {
-      ...hwFields(), hardware_source: "detected", webgpu_score: hw.quick_score,
-      benchmark_type: hw.quick_score != null ? "webgpu_quick" : null,
-      benchmark_version: hw.quick_score != null ? window.HW.BENCH_VERSION : null,
-      benchmark_duration_ms: hw.duration_ms,
-    });
     explore({ reason: "filter" });
   }
 
@@ -697,7 +730,7 @@
     try {
       const res = await window.HW.fullBenchmark((p) => { prog.value = p; });
       $("#bench-out").textContent = `WebGPU compute score: ${res.score.toLocaleString()}. A relative browser GPU score for grouping hardware, not MLX speed or tokens per second.`;
-      Telemetry.track("browser_benchmark", {
+      Telemetry.contribute("browser_benchmark", {
         ...hwFields(), webgpu_score: res.score, benchmark_type: "webgpu_full", benchmark_version: res.version,
         benchmark_duration_ms: res.duration_ms, gpu_capability_class: window.HW.capabilityClass(state.hw && state.hw.quick_score),
       });
@@ -742,7 +775,7 @@
     if (!obj || typeof obj !== "object" || Array.isArray(obj)) { out.textContent = "Paste a single JSON object, the block the script printed."; return; }
     const fields = {};
     for (const k of BENCH_KEYS) if (k in obj) fields[k] = obj[k];
-    const res = await Telemetry.submit("mlx_benchmark_submission", fields);
+    const res = await Telemetry.submit("mlx_benchmark", fields);
     out.textContent = res.ok
       ? res.flags.length ? `Result recorded and flagged for review: ${res.flags.join(", ")}.` : "Result recorded anonymously. Thank you."
       : res.message;

static/index.html

@@ -151,7 +151,7 @@
       </div>
       <details>
         <summary>Paste a result you already ran</summary>
-        <textarea id="bench-json" rows="6" maxlength="4000" spellcheck="false" placeholder='{"event_type": "mlx_benchmark_submission", ...}'></textarea>
+        <textarea id="bench-json" rows="6" maxlength="4000" spellcheck="false" placeholder='{"event_type": "mlx_benchmark", ...}'></textarea>
         <button type="button" id="bench-submit" class="solid">Submit result</button>
         <p id="bench-submit-out" class="fine" aria-live="polite"></p>
       </details>
@@ -194,7 +194,7 @@
       <h2 id="priv-h">Privacy and data</h2>
       <p>We collect anonymous model-selection and optional benchmark data to improve MLX Model Explorer and community recommendations. It's published as the <a id="dataset-link" href="https://huggingface.co/datasets/mlx-community/mlx-model-explorer-data" target="_blank" rel="noopener">mlx-model-explorer-data</a> dataset.</p>
       <ul>
-        <li><strong>Collected:</strong> the filters you choose, models you open or compare, coarse hardware class (GPU vendor and architecture, core count, browser and OS family), test scores, and feedback you send.</li>
+        <li><strong>Collected:</strong> one summary per visit (the filters you ended on, the families and quantizations you looked at, which models you opened, compared or visited, and a coarse hardware class: memory, GPU vendor and architecture, core count, browser and OS family), plus test scores and feedback you choose to send. Individual clicks aren't recorded.</li>
         <li><strong>Never collected:</strong> name, email, IP address, location, cookies, user-agent strings or device fingerprints. Links, emails and phone numbers are stripped from notes.</li>
         <li><strong>Session ID:</strong> random, lives only in this tab, and is gone when you close it.</li>
         <li>The <a href="/stats">stats page</a> only shows groups of five or more.</li>

static/stats.html

@@ -24,11 +24,11 @@
 <main class="shell">
   <section class="stats-hero">
     <h1>What the MLX community is trying to run</h1>
-    <p class="lede">Anonymous totals from MLX Model Explorer. Groups smaller than <span id="kmin">5</span> are combined into "other", flagged submissions are left out, and no individual record is shown.</p>
+    <p class="lede">Anonymous totals from MLX Model Explorer. Each visit counts once. Groups smaller than <span id="kmin">5</span> are combined into "other", flagged submissions are left out, and no individual record is shown.</p>
   </section>
   <div class="totals" id="totals"></div>
   <section class="chart" aria-labelledby="q-h">
-    <h2 id="q-h">Quantization people search for</h2>
+    <h2 id="q-h">Quantizations people look at</h2>
     <div class="qshare" id="qshare" role="img" aria-label="Quantization share"></div>
     <p class="scale-note" id="qshare-note"></p>
     <p class="scale-note" id="avgctx"></p>

static/stats.js

@@ -9,13 +9,12 @@
     return n;
   }
   const LABELS = {
-    events: "Events", sessions: "Sessions", searches: "Searches", distinct_queries: "Distinct queries", model_views: "Models opened",
-    model_clicks: "Hugging Face visits", hardware_tests: "Hardware checks", browser_benchmarks: "GPU tests",
-    feedback: "Reports", mlx_benchmarks: "MLX benchmarks",
+    sessions: "Visits", engaged_sessions: "Visits that opened a model", hardware_checks: "Hardware checks",
+    browser_benchmarks: "GPU tests", feedback: "Reports", mlx_benchmarks: "MLX benchmarks",
   };
   const CHARTS = [
     ["families", "Model families"], ["sizes", "Model sizes"], ["ram_classes", "Mac memory (GB)"],
-    ["contexts", "Target context"], ["priorities", "What matters most"], ["selected_models", "Most compared and visited models"],
+    ["contexts", "Target context"], ["priorities", "What matters most"], ["selected_models", "Most compared or visited models"],
     ["gpu_capability", "Browser GPU class"], ["webgpu_available", "WebGPU available"],
   ];
   const QCOLORS = { "2-bit": "#8a4f9e", "3-bit": "#7a5ab8", "4-bit": "#2f5d8c", "5-bit": "#3f7f86", "6-bit": "#2e7d4f", "8-bit": "#62802a" };
@@ -68,7 +67,7 @@
       seg.title = `${k}: ${Math.round(v * 100)}%`;
       return seg;
     }));
-    $("#qshare-note").textContent = q.length ? q.map(([k, v]) => `${k} ${Math.round(v * 100)}%`).join(", ") : "Not enough searches with a quantization filter yet.";
+    $("#qshare-note").textContent = q.length ? q.map(([k, v]) => `${k} ${Math.round(v * 100)}%`).join(", ") : "Not enough visits with a quantization filter yet.";
     $("#avgctx").textContent = s.average_target_context ? `Average target context: ${Math.round(s.average_target_context / 1024)}k tokens.` : "";
     $("#charts").replaceChildren(...CHARTS.map(([key, title]) => {
       const c = el("section", "chart");

tests/test_api.py

@@ -110,22 +110,28 @@ def test_model_detail(client):
 def test_events_ingest_to_parquet(client, tmp_path):
     sid = "abcdef0123456789"
     batch = {"events": [
-        {"event_type": "search", "session_id": sid, "model_family": "Qwen", "quantization": "4-bit",
-         "target_context": 32768, "priority": "balanced", "hardware_source": "confirmed", "hardware_memory_class": 36},
-        {"event_type": "model_select", "session_id": sid, "selected_model": "mlx-community/Qwen3-8B-4bit",
-         "selected_model_rank": 0, "target_context": 32768, "hardware_memory_class": 36},
+        {"event_type": "session", "session_id": sid, "model_family": "Qwen", "quantization": "4-bit",
+         "target_context": 32768, "priority": "balanced", "hardware_source": "confirmed", "hardware_memory_class": 36,
+         "families_searched": ["Qwen"], "quants_searched": ["4-bit"], "distinct_queries": 2,
+         "models_viewed": ["mlx-community/Qwen3-8B-4bit"]},
         {"event_type": "feedback", "session_id": sid, "selected_model": "mlx-community/Qwen3-8B-4bit",
          "tried": "yes", "quality_rating": "good", "notes": "worked fine, email me x@y.com"},
+        {"event_type": "search", "session_id": sid, "model_family": "Qwen"},
     ]}
     r = client.post("/api/events", json=batch)
-    assert r.status_code == 202 and r.json()["accepted"] == 3
+    assert r.status_code == 202 and r.json()["accepted"] == 2
+    assert r.json()["flags"][2] == ["dropped_legacy_event"]
+    # the visit re-sends its summary later: the newest row wins in stats
+    client.post("/api/events", json={"events": [dict(batch["events"][0], distinct_queries=5,
+                                                     models_compared=["mlx-community/Qwen3-14B-4bit"])]})
     assert client.st.sink.flush()
     rows = client.st.sink.read_existing()
-    assert [x["event_type"] for x in rows] == ["search", "model_select", "feedback"]
-    assert rows[1]["hf_downloads_at_selection"] == 90000
-    assert "x@y.com" not in rows[2]["notes"]
+    assert [x["event_type"] for x in rows] == ["session", "feedback", "session"]
+    assert rows[1]["hf_downloads_at_selection"] == 90000 and "x@y.com" not in rows[1]["notes"]
+    client.st.stats.invalidate()
     stats = client.get("/api/stats").json()
-    assert stats["totals"]["events"] == 3 and stats["totals"]["sessions"] == 1
+    assert stats["totals"]["sessions"] == 1 and stats["totals"]["engaged_sessions"] == 1
+    assert stats["totals"]["feedback"] == 1
     assert stats["families"] == [{"key": "other", "count": 1, "share": 1.0}]  # k-anonymity folds tiny buckets
 
 
@@ -166,19 +172,21 @@ def test_collection_disabled(client):
     assert r.json() == {"accepted": 0} and client.st.sink.pending == 0
 
 
-def test_stats_dedupe_and_suppression():
+def test_stats_one_count_per_visit_and_suppression():
     from app.stats import compute
     rows = []
     for i in range(6):
         sid = f"{i:016x}"
-        for _ in range(3):  # the same query re-rendered three times counts once
-            rows.append({"event_type": "filter", "session_id": sid, "model_family": "Qwen", "quantization": "4-bit",
-                         "target_context": 8192, "hardware_memory_class": 36, "suspicious_flags": []})
-    rows.append({"event_type": "search", "session_id": "ffffffffffffffff", "model_family": "Rare",
-                 "quantization": "3-bit", "suspicious_flags": []})
-    rows.append({"event_type": "search", "model_family": "Flagged", "suspicious_flags": ["unknown_model"]})
+        for n in range(3):  # each visit re-sent its summary three times
+            rows.append({"event_type": "session", "session_id": sid, "timestamp": f"2026-09-14T01:0{n}:00Z",
+                         "families_searched": ["Qwen"], "quants_searched": ["4-bit"], "target_context": 8192,
+                         "hardware_memory_class": 36, "suspicious_flags": []})
+    rows.append({"event_type": "session", "session_id": "f" * 16, "timestamp": "2026-09-14T02:00:00Z",
+                 "families_searched": ["Rare"], "quants_searched": ["3-bit"], "suspicious_flags": []})
+    rows.append({"event_type": "session", "session_id": "e" * 16, "families_searched": ["Flagged"],
+                 "suspicious_flags": ["unknown_model_in_session"]})
     s = compute(rows)
-    assert s["totals"]["distinct_queries"] == 7
+    assert s["totals"]["sessions"] == 7
     assert s["families"] == [{"key": "Qwen", "count": 6, "share": 0.8571}, {"key": "other", "count": 1, "share": 0.1429}]
     assert s["ram_classes"][0] == {"key": 36, "count": 6, "share": 1.0}
-    assert all(f["key"] != "Flagged" for f in s["families"])
+    assert s["quant_share"]["4-bit"] == 0.8571

tests/test_compact.py

@@ -0,0 +1,43 @@
+import importlib.util
+from pathlib import Path
+
+from app.events import COLUMNS
+
+spec = importlib.util.spec_from_file_location("compact", Path(__file__).resolve().parent.parent / "scripts" / "compact.py")
+compact = importlib.util.module_from_spec(spec)
+spec.loader.exec_module(compact)
+
+
+def test_plan_selects_only_that_month():
+    files = [
+        "README.md",
+        "data/incoming/2026/09/01/a-1.parquet", "data/incoming/2026/09/30/b-2.parquet",
+        "data/incoming/2026/10/01/c-3.parquet", "data/events/2026-08.parquet", "data/events/2026-09.parquet",
+    ]
+    incoming, existing, target = compact.plan(files, "2026-09")
+    assert incoming == ["data/incoming/2026/09/01/a-1.parquet", "data/incoming/2026/09/30/b-2.parquet"]
+    assert existing == "data/events/2026-09.parquet" and target == "data/events/2026-09.parquet"
+    assert compact.plan(files, "2026-11") == ([], None, "data/events/2026-11.parquet")
+
+
+def test_compact_rows_merges_aligns_and_dedupes():
+    old_monthly = [{"event_type": "session", "session_id": "a" * 16, "timestamp": "2026-09-02T00:00:00Z",
+                    "distinct_queries": 1}]
+    shard1 = [{"event_type": "session", "session_id": "a" * 16, "timestamp": "2026-09-02T00:10:00Z",
+               "distinct_queries": 3},
+              {"event_type": "mlx_benchmark", "selected_model": "mlx-community/x-4bit", "timestamp": "2026-09-02T00:11:00Z"}]
+    shard2 = [{"event_type": "session", "session_id": "b" * 16, "timestamp": "2026-09-03T00:00:00Z"}]
+    rows = compact.compact_rows([old_monthly, shard1, shard2])
+    assert len(rows) == 3 and all(set(r) == set(COLUMNS) for r in rows)
+    assert [r["distinct_queries"] for r in rows if r["session_id"] == "a" * 16] == [3]
+    assert [r["timestamp"] for r in rows] == sorted(r["timestamp"] for r in rows)
+
+
+def test_previous_month_and_refuses_current():
+    from datetime import datetime, timezone
+    assert compact.previous_month(datetime(2026, 1, 15, tzinfo=timezone.utc)) == "2025-12"
+    assert compact.previous_month(datetime(2026, 9, 14, tzinfo=timezone.utc)) == "2026-08"
+    import pytest
+    now = datetime.now(timezone.utc).strftime("%Y-%m")
+    with pytest.raises(SystemExit):
+        compact.main(["--month", now])

tests/test_events.py

@@ -24,26 +24,65 @@ def ev(**kw):
     return ClientEvent(**{"event_type": "search", "session_id": SID, **kw})
 
 
-def test_valid_search_row_has_all_columns_and_no_pii():
-    row = to_row(ev(model_family="Qwen", parameter_bucket="8-15B", quantization="4-bit",
+def test_session_row_has_all_columns_and_no_pii():
+    row = to_row(ev(event_type="session", model_family="Qwen", parameter_bucket="8-15B", quantization="4-bit",
                     target_context=32768, priority="balanced", hardware_source="confirmed",
-                    hardware_memory_class=36), CAT)
+                    hardware_memory_class=36, families_searched=["Qwen", "Gemma", "Qwen"], quants_searched=["4-bit"],
+                    distinct_queries=3, top_model="mlx-community/Qwen3-8B-4bit", top_model_score=88.2,
+                    top_model_fit="Comfortable", models_viewed=["mlx-community/Qwen3-8B-4bit"]), CAT)
     assert set(row) == set(COLUMNS)
+    assert row["schema_version"] == 2 and row["event_type"] == "session"
+    assert row["families_searched"] == ["Qwen", "Gemma"]  # de-duplicated
     assert row["hardware_confirmed"] is True and row["suspicious_flags"] == []
     assert row["timestamp"].endswith("Z") and "." not in row["timestamp"]
-    for forbidden in ("ip", "user_agent", "email", "name", "cookie"):
+    for forbidden in ("ip", "user_agent", "email", "cookie"):
         assert forbidden not in row
 
 
-def test_server_fills_catalogue_facts():
-    row = to_row(ev(event_type="model_select", selected_model="mlx-community/Qwen3-8B-4bit",
-                    selected_model_rank=0), CAT)
-    assert row["hf_downloads_at_selection"] == 42 and row["quant_bits"] == 4
+def test_feedback_gets_catalogue_facts():
+    row = to_row(ev(event_type="feedback", selected_model="mlx-community/Qwen3-8B-4bit", tried="yes",
+                    quality_rating="good"), CAT)
+    assert row["hf_downloads_at_selection"] == 42 and row["quant_bits"] == 4 and row["outcome"] == "worked"
     assert row["model_family"] == "Qwen" and row["parameter_bucket"] == "8-15B"
 
 
+def test_legacy_clickstream_events_are_dropped_and_old_bench_name_is_accepted():
+    assert to_row(ev(event_type="search", model_family="Qwen"), CAT) is None
+    assert to_row(ev(event_type="model_view", selected_model="mlx-community/Qwen3-8B-4bit"), CAT) is None
+    row = to_row(ev(event_type="mlx_benchmark", selected_model="mlx-community/Qwen3-8B-4bit",
+                    generation_tps=40, benchmark_version="mlxbench-1"), CAT)
+    assert row["event_type"] == "mlx_benchmark" and row["suspicious_flags"] == []
+
+
+def test_session_lists_are_capped_and_checked():
+    with pytest.raises(ValidationError):
+        ev(event_type="session", models_viewed=[f"mlx-community/m{i}" for i in range(11)])
+    with pytest.raises(ValidationError):
+        ev(event_type="session", models_compared=["../etc/passwd"])
+    with pytest.raises(ValidationError):
+        ev(event_type="session", families_searched=["a", "b", "c", "d", "e", "f"])
+    row = to_row(ev(event_type="session", models_clicked=["mlx-community/Nope-1B"]), CAT)
+    assert "unknown_model_in_session" in row["suspicious_flags"]
+
+
+def test_latest_sessions_keeps_newest_row_per_visit():
+    from app.events import latest_sessions
+    rows = [
+        {"event_type": "session", "session_id": "a" * 16, "timestamp": "2026-09-14T01:00:00Z", "distinct_queries": 1},
+        {"event_type": "feedback", "session_id": "a" * 16, "timestamp": "2026-09-14T01:00:30Z"},
+        {"event_type": "session", "session_id": "a" * 16, "timestamp": "2026-09-14T01:05:00Z", "distinct_queries": 4},
+        {"event_type": "session", "session_id": "b" * 16, "timestamp": "2026-09-14T01:05:00Z", "distinct_queries": 1},
+        {"event_type": "session", "session_id": "b" * 16, "timestamp": "2026-09-14T01:05:00Z", "distinct_queries": 2},
+    ]
+    out = latest_sessions(rows)
+    sessions = {r["session_id"]: r["distinct_queries"] for r in out if r["event_type"] == "session"}
+    assert sessions == {"a" * 16: 4, "b" * 16: 2}  # later arrival wins a same-second tie
+    assert sum(1 for r in out if r["event_type"] == "feedback") == 1 and len(out) == 3
+
+
 @pytest.mark.parametrize("bad", [
     {"event_type": "delete_everything"},
+    {"event_type": "session", "compare_models": ["a/b"]},
     {"event_type": "search", "session_id": "NOT-HEX"},
     {"event_type": "search", "target_context": 12345},
     {"event_type": "search", "hardware_memory_class": 17},
@@ -56,7 +95,6 @@ def test_server_fills_catalogue_facts():
     {"event_type": "search", "ip": "1.2.3.4"},
     {"event_type": "search", "model_family": "<img src=x>"},
     {"event_type": "search", "benchmark_version": "1.0; rm -rf"},
-    {"event_type": "search", "compare_models": ["a/b"] * 6},
 ])
 def test_rejects_malformed(bad):
     with pytest.raises(ValidationError):
@@ -67,7 +105,7 @@ def test_batch_limits():
     with pytest.raises(ValidationError):
         EventBatch(events=[])
     with pytest.raises(ValidationError):
-        EventBatch(events=[{"event_type": "search"}] * 51)
+        EventBatch(events=[{"event_type": "session"}] * 21)
 
 
 def test_notes_are_scrubbed_and_capped():
@@ -80,20 +118,20 @@ def test_notes_are_scrubbed_and_capped():
 
 
 def test_flags_unknown_model_and_implausible_tps():
-    row = to_row(ev(event_type="mlx_benchmark_submission", selected_model="mlx-community/Nope-1B",
+    row = to_row(ev(event_type="mlx_benchmark", selected_model="mlx-community/Nope-1B",
                     generation_tps=10, benchmark_version="1", benchmark_type="mlx_lm"), CAT)
     assert "unknown_model" in row["suspicious_flags"]
-    row = to_row(ev(event_type="mlx_benchmark_submission", selected_model="mlx-community/Llama-3.3-70B-Instruct-4bit",
+    row = to_row(ev(event_type="mlx_benchmark", selected_model="mlx-community/Llama-3.3-70B-Instruct-4bit",
                     generation_tps=1500, benchmark_version="1", benchmark_type="mlx_lm"), CAT)
     assert "implausible_tps" in row["suspicious_flags"]
-    row = to_row(ev(event_type="mlx_benchmark_submission", selected_model="mlx-community/Qwen3-8B-4bit",
+    row = to_row(ev(event_type="mlx_benchmark", selected_model="mlx-community/Qwen3-8B-4bit",
                     generation_tps=45, benchmark_version="1", benchmark_type="mlx_lm", peak_memory_gb=5.1,
                     reported_ram_gb=36), CAT)
     assert row["suspicious_flags"] == []
 
 
 def test_flags_other_inconsistencies():
-    row = to_row(ev(event_type="mlx_benchmark_submission", peak_memory_gb=40, reported_ram_gb=16), CAT)
+    row = to_row(ev(event_type="mlx_benchmark", peak_memory_gb=40, reported_ram_gb=16), CAT)
     assert {"peak_memory_exceeds_ram", "incomplete_benchmark"} <= set(row["suspicious_flags"])
     row = to_row(ev(event_type="feedback", tried="yes", quality_rating="good", failure_reason="too_slow"), CAT)
     assert "conflicting_feedback" in row["suspicious_flags"]
@@ -103,17 +141,17 @@ def test_flags_other_inconsistencies():
 
 
 def test_quality_submission_validation():
-    ok = to_row(ev(event_type="mlx_benchmark_submission", selected_model="mlx-community/Qwen3-8B-4bit",
+    ok = to_row(ev(event_type="mlx_benchmark", selected_model="mlx-community/Qwen3-8B-4bit",
                    benchmark_type="mlx_lm", benchmark_version="mlxbench-1", perplexity=9.8, perplexity_stderr=0.2,
                    eval_dataset="wikitext2-test-v1", eval_tokens=16368), CAT)
     assert ok["suspicious_flags"] == [] and ok["perplexity"] == 9.8
-    bad = to_row(ev(event_type="mlx_benchmark_submission", selected_model="mlx-community/Qwen3-8B-4bit",
+    bad = to_row(ev(event_type="mlx_benchmark", selected_model="mlx-community/Qwen3-8B-4bit",
                     benchmark_version="mlxbench-1", perplexity=5000, eval_dataset="wikitext2-test-v1", eval_tokens=16368), CAT)
     assert "implausible_perplexity" in bad["suspicious_flags"]
-    inc = to_row(ev(event_type="mlx_benchmark_submission", selected_model="mlx-community/Qwen3-8B-4bit",
+    inc = to_row(ev(event_type="mlx_benchmark", selected_model="mlx-community/Qwen3-8B-4bit",
                     benchmark_version="mlxbench-1", perplexity=9.8), CAT)
     assert "incomplete_quality" in inc["suspicious_flags"]
     with pytest.raises(ValidationError):
-        ev(event_type="mlx_benchmark_submission", perplexity=0.5)
+        ev(event_type="mlx_benchmark", perplexity=0.5)
     with pytest.raises(ValidationError):
-        ev(event_type="mlx_benchmark_submission", perplexity=9.8, eval_dataset="my-own-text")
+        ev(event_type="mlx_benchmark", perplexity=9.8, eval_dataset="my-own-text")

tests/test_sink.py

@@ -6,7 +6,7 @@ from app.sink import HubParquetSink, LocalParquetSink, rows_to_parquet_bytes
 
 
 def rows(n):
-    return [to_row(ClientEvent(event_type="search", model_family="Qwen", quantization="4-bit")) for _ in range(n)]
+    return [to_row(ClientEvent(event_type="session", model_family="Qwen", quantization="4-bit")) for _ in range(n)]
 
 
 def test_batch_becomes_one_shard(tmp_path):
@@ -14,14 +14,14 @@ def test_batch_becomes_one_shard(tmp_path):
     s.add(rows(7))
     s.add(rows(5))
     assert s.flush()
-    shards = list((tmp_path / "data/events").rglob("*.parquet"))
+    shards = list((tmp_path / "data/incoming").rglob("*.parquet"))
     assert len(shards) == 1
     t = pq.read_table(shards[0])
     assert t.num_rows == 12 and t.column_names == list(COLUMNS)
-    assert s.flush() and len(list((tmp_path / "data/events").rglob("*.parquet"))) == 1  # empty flush = no shard
+    assert s.flush() and len(list((tmp_path / "data/incoming").rglob("*.parquet"))) == 1  # empty flush = no shard
     s.add(rows(1))
     s.flush()
-    assert len(list((tmp_path / "data/events").rglob("*.parquet"))) == 2  # appends, never rewrites
+    assert len(list((tmp_path / "data/incoming").rglob("*.parquet"))) == 2  # appends, never rewrites
     assert len(s.read_existing()) == 13
 
 
@@ -50,7 +50,7 @@ def test_hub_failure_keeps_spool_and_retries(tmp_path):
     assert not s2.flush()
     assert s2.flush() and s2.pending == 0 and s2.status()["healthy"]
     assert len(api.commits) == 1 and len(api.commits[0]) == 1
-    assert api.commits[0][0].startswith("data/events/") and api.commits[0][0].endswith(".parquet")
+    assert api.commits[0][0].startswith("data/incoming/") and api.commits[0][0].endswith(".parquet")
     s3 = HubParquetSink("org/data", None, spool_dir=tmp_path, api=api, flush_seconds=999)
     assert s3.pending == 0
 
@@ -58,13 +58,38 @@ def test_hub_failure_keeps_spool_and_retries(tmp_path):
 def test_local_refuses_overwrite(tmp_path):
     s = LocalParquetSink(tmp_path, flush_seconds=999)
     data = rows_to_parquet_bytes(rows(1))
-    s._write_shard("data/events/x.parquet", data)
+    s._write_shard("data/incoming/x.parquet", data)
     with pytest.raises(FileExistsError):
-        s._write_shard("data/events/x.parquet", data)
+        s._write_shard("data/incoming/x.parquet", data)
 
 
 def test_dataset_schema_is_deliberate():
     """Changing columns changes the public dataset schema: bump this list on purpose, and
     migrate or document existing shards (see the dataset card's versioning section)."""
-    assert len(COLUMNS) == 62
+    assert len(COLUMNS) == 67
     assert {"perplexity", "perplexity_stderr", "eval_dataset", "eval_tokens"} <= set(COLUMNS)
+
+
+def test_size_triggered_flush(tmp_path, monkeypatch):
+    import time
+    import app.sink as sink_mod
+    monkeypatch.setattr(sink_mod, "FLUSH_ROWS", 5)
+    s = LocalParquetSink(tmp_path, flush_seconds=999)
+    s.add(rows(3))
+    assert s.pending == 3
+    s.add(rows(3))  # crosses FLUSH_ROWS -> flushes in the background
+    for _ in range(50):
+        if s.pending == 0:
+            break
+        time.sleep(0.05)
+    assert s.pending == 0 and len(list((tmp_path / "data/incoming").rglob("*.parquet"))) == 1
+
+
+def test_reads_monthly_and_incoming(tmp_path):
+    from app.sink import rows_to_parquet_bytes as rb
+    (tmp_path / "data/events").mkdir(parents=True)
+    (tmp_path / "data/events/2026-08.parquet").write_bytes(rb(rows(4)))
+    s = LocalParquetSink(tmp_path, flush_seconds=999)
+    s.add(rows(2))
+    s.flush()
+    assert len(s.read_existing()) == 6