diff --git a/experiments/for_you.py b/experiments/for_you.py
new file mode 100644
index 0000000..94670a1
--- /dev/null
+++ b/experiments/for_you.py
@@ -0,0 +1,505 @@
+#!/usr/bin/env python3
+"""
+For You — personalized Bluesky feed from the Divepool firehose.
+
+Uses ~15 medoid centroid embeddings from search API clusters.
+Window-based selection: scores every firehose post, picks the best per window.
+Includes detailed filter stats (for_you_stats.json).
+
+Usage:
+    python3 for_you.py <token> <did> --window 60 [--top 3] [--floor 0.5]
+"""
+
+import argparse
+import json
+import os
+import sys
+import threading
+import time
+import urllib.error
+import urllib.parse
+import urllib.request
+from collections import defaultdict
+
+import numpy as np
+
+DIVEPOOL_SEARCH = "https://divepool.social/api/v1/search"
+DIVEPOOL_STREAM = "https://divepool.social/api/v1/embeddings"
+BSKY_API = "https://public.api.bsky.app/xrpc"
+STATS_FILE = os.path.join(os.path.dirname(os.path.abspath(__file__)), "for_you_stats.json")
+
+
+# ── API helpers ──────────────────────────────────────────────────────────────
+
+def divepool_search(token, query, limit=100, did=None, cluster=False, distinct=True):
+    payload = {"query": query, "limit": limit, "distinct": distinct}
+    if did:
+        payload["did"] = did
+    if cluster:
+        payload["cluster"] = True
+    data = json.dumps(payload).encode()
+    req = urllib.request.Request(DIVEPOOL_SEARCH, data=data, headers={
+        "Content-Type": "application/json",
+        "Authorization": f"Bearer {token}",
+    }, method="POST")
+    with urllib.request.urlopen(req, timeout=15) as resp:
+        return json.loads(resp.read())
+
+
+def resolve_post_text(did, collection, rkey):
+    uri = f"at://{did}/{collection}/{rkey}"
+    url = f"{BSKY_API}/app.bsky.feed.getPostThread?uri={urllib.parse.quote(uri)}&depth=0"
+    req = urllib.request.Request(url, headers={"Accept": "application/json"})
+    try:
+        with urllib.request.urlopen(req, timeout=5) as resp:
+            data = json.loads(resp.read())
+            post = data.get("thread", {}).get("post", {})
+            text = post.get("record", {}).get("text", "")
+            handle = post.get("author", {}).get("handle", "")
+            return handle, text
+    except Exception:
+        return None, None
+
+
+def bsky_link(handle, rkey):
+    return f"https://bsky.app/profile/{handle}/post/{rkey}"
+
+
+def tid_to_timestamp(tid):
+    """Decode an AT Protocol TID (base32-sortable) to a Unix timestamp in seconds.
+    TIDs encode microseconds-since-epoch in the high 53 bits."""
+    charset = "234567abcdefghijklmnopqrstuvwxyz"
+    try:
+        n = 0
+        for ch in tid:
+            n = n * 32 + charset.index(ch)
+        # high 53 bits = microseconds since epoch
+        usec = n >> 10
+        return usec / 1_000_000
+    except (ValueError, IndexError):
+        return 0.0
+
+
+# ── Firehose streaming ──────────────────────────────────────────────────────
+
+def stream_firehose(token, post_queue, stop_event):
+    import zstandard as zstd
+    while not stop_event.is_set():
+        try:
+            req = urllib.request.Request(DIVEPOOL_STREAM, headers={
+                "Authorization": f"Bearer {token}",
+            })
+            with urllib.request.urlopen(req, timeout=30) as resp:
+                dctx = zstd.ZstdDecompressor()
+                reader = dctx.stream_reader(resp)
+                buf = b""
+                while not stop_event.is_set():
+                    chunk = reader.read(8192)
+                    if not chunk:
+                        break
+                    buf += chunk
+                    while b"\n" in buf:
+                        line, buf = buf.split(b"\n", 1)
+                        if not line.strip():
+                            continue
+                        try:
+                            batch = json.loads(line)
+                        except json.JSONDecodeError:
+                            continue
+                        if not isinstance(batch, dict):
+                            continue
+                        dids = batch.get("did")
+                        if not isinstance(dids, list) or not dids:
+                            continue
+                        cols = batch.get("col", [])
+                        rkeys = batch.get("rkey", [])
+                        langs = batch.get("lang", [])
+                        cs = batch.get("c", [])
+                        for i in range(len(dids)):
+                            col = cols[i] if i < len(cols) else ""
+                            if "feed.post" not in col:
+                                continue
+                            c_emb = cs[i] if i < len(cs) else []
+                            if not c_emb:
+                                continue
+                            rkey = rkeys[i] if i < len(rkeys) else ""
+                            lang = langs[i] if i < len(langs) else ""
+                            post_queue.append((dids[i], col, rkey, lang, c_emb))
+        except Exception:
+            if not stop_event.is_set():
+                time.sleep(2)
+
+
+# ── Stats tracker ──────────────────────────────────────────────────────────
+
+class FilterStats:
+    """Tracks scoring distribution and selection stats."""
+
+    BINS = [0.0, 0.3, 0.4, 0.5, 0.55, 0.6, 0.65, 0.7, 0.75, 0.8, 0.85, 0.9, 0.95, 1.0]
+
+    def __init__(self):
+        self.started = time.time()
+        self.total = 0
+        self.skip_seen_self = 0
+        self.fail_resolve = 0
+        self.fail_spam = 0
+        self.shown = 0
+        self.windows_elapsed = 0
+        self.windows_empty = 0        # windows where nothing was good enough
+        # histogram of composite scores for all scored posts
+        self.score_hist = np.zeros(len(self.BINS) - 1, dtype=np.int64)
+        # histogram of shown post scores
+        self.shown_hist = np.zeros(len(self.BINS) - 1, dtype=np.int64)
+        # per-medoid hit counts (best medoid for shown posts)
+        self.medoid_hits = defaultdict(int)
+        # track recent shown timestamps for rate calc
+        self._shown_times = []
+        # recency: track age of incoming posts
+        self._recent_ages = []
+
+    def record_age(self, rkey):
+        ts = tid_to_timestamp(rkey)
+        if ts > 0:
+            age = time.time() - ts
+            self._recent_ages.append((time.time(), age))
+
+    def age_stats(self):
+        now = time.time()
+        cutoff = now - 300
+        self._recent_ages = [(t, a) for t, a in self._recent_ages if t > cutoff]
+        if not self._recent_ages:
+            return None
+        ages = [a for _, a in self._recent_ages]
+        return {
+            "count": len(ages),
+            "median_sec": round(float(np.median(ages)), 1),
+            "p90_sec": round(float(np.percentile(ages, 90)), 1),
+            "max_sec": round(max(ages), 1),
+            "min_sec": round(min(ages), 1),
+            "pct_under_60s": round(100 * sum(1 for a in ages if a < 60) / len(ages), 1),
+            "pct_under_300s": round(100 * sum(1 for a in ages if a < 300) / len(ages), 1),
+        }
+
+    def record_score(self, score):
+        idx = np.searchsorted(self.BINS, score, side="right") - 1
+        idx = max(0, min(idx, len(self.score_hist) - 1))
+        self.score_hist[idx] += 1
+
+    def record_shown(self, score):
+        self.shown += 1
+        self._shown_times.append(time.time())
+        idx = np.searchsorted(self.BINS, score, side="right") - 1
+        idx = max(0, min(idx, len(self.shown_hist) - 1))
+        self.shown_hist[idx] += 1
+
+    def shown_per_min(self):
+        now = time.time()
+        cutoff = now - 300
+        self._shown_times = [t for t in self._shown_times if t > cutoff]
+        if not self._shown_times:
+            return 0.0
+        window = now - self._shown_times[0]
+        if window < 10:
+            return 0.0
+        return len(self._shown_times) / (window / 60)
+
+    def _hist_to_dict(self, hist):
+        out = {}
+        for i, count in enumerate(hist):
+            if count > 0:
+                lo, hi = self.BINS[i], self.BINS[i + 1]
+                out[f"{lo:.2f}-{hi:.2f}"] = int(count)
+        return out
+
+    def to_dict(self, floor):
+        elapsed = time.time() - self.started
+        scored = self.total - self.skip_seen_self
+        return {
+            "elapsed_min": round(elapsed / 60, 1),
+            "floor": round(floor, 4),
+            "shown_per_min": round(self.shown_per_min(), 2),
+            "counts": {
+                "total": self.total,
+                "skip_seen_self": self.skip_seen_self,
+                "scored": scored,
+                "fail_resolve": self.fail_resolve,
+                "fail_spam": self.fail_spam,
+                "shown": self.shown,
+                "windows": self.windows_elapsed,
+                "windows_empty": self.windows_empty,
+            },
+            "rates": {
+                "shown_pct": round(100 * self.shown / scored, 4) if scored else 0,
+                "window_hit_pct": round(100 * (self.windows_elapsed - self.windows_empty)
+                                        / self.windows_elapsed, 1) if self.windows_elapsed else 0,
+            },
+            "score_histogram": self._hist_to_dict(self.score_hist),
+            "shown_histogram": self._hist_to_dict(self.shown_hist),
+            "medoid_hits": dict(sorted(self.medoid_hits.items(), key=lambda x: -x[1])),
+            "recency": self.age_stats(),
+        }
+
+    def write_file(self, floor):
+        data = self.to_dict(floor)
+        tmp = STATS_FILE + ".tmp"
+        with open(tmp, "w") as f:
+            json.dump(data, f, indent=2)
+            f.write("\n")
+        os.replace(tmp, STATS_FILE)
+
+    def print_summary(self, floor):
+        d = self.to_dict(floor)
+        c = d["counts"]
+        r = d["rates"]
+        print(f"\n{'='*60}", file=sys.stderr)
+        print(f"Filter stats  ({d['elapsed_min']} min, floor={d['floor']})",
+              file=sys.stderr)
+        print(f"{'='*60}", file=sys.stderr)
+        print(f"  Total scanned:     {c['total']:>8}", file=sys.stderr)
+        print(f"  Skip seen/self:    {c['skip_seen_self']:>8}", file=sys.stderr)
+        print(f"  Scored:            {c['scored']:>8}", file=sys.stderr)
+        print(f"  Fail resolve:      {c['fail_resolve']:>8}", file=sys.stderr)
+        print(f"  Fail spam:         {c['fail_spam']:>8}", file=sys.stderr)
+        print(f"  Shown:             {c['shown']:>8}  "
+              f"({r['shown_pct']:.3f}% of scored)", file=sys.stderr)
+        print(f"  Windows:           {c['windows']:>8}  "
+              f"({c['windows_empty']} empty, {r['window_hit_pct']:.0f}% hit)",
+              file=sys.stderr)
+        print(f"\n  Shown/min (5m window): {d['shown_per_min']}", file=sys.stderr)
+        if d["score_histogram"]:
+            print(f"\n  Score distribution (all scored posts):", file=sys.stderr)
+            for bucket, count in d["score_histogram"].items():
+                bar = "#" * min(count, 60)
+                print(f"    {bucket}: {count:>7}  {bar}", file=sys.stderr)
+        if d["shown_histogram"]:
+            print(f"\n  Shown post scores:", file=sys.stderr)
+            for bucket, count in d["shown_histogram"].items():
+                bar = "#" * min(count, 60)
+                print(f"    {bucket}: {count:>7}  {bar}", file=sys.stderr)
+        if d["medoid_hits"]:
+            print(f"\n  Medoid hits (shown posts):", file=sys.stderr)
+            for label, count in d["medoid_hits"].items():
+                print(f"    {count:>6}  {label}", file=sys.stderr)
+        rec = d.get("recency")
+        if rec:
+            print(f"\n  Post recency (5m window, {rec['count']} posts):", file=sys.stderr)
+            print(f"    Median age:   {rec['median_sec']:.0f}s", file=sys.stderr)
+            print(f"    P90 age:      {rec['p90_sec']:.0f}s", file=sys.stderr)
+            print(f"    Range:        {rec['min_sec']:.0f}s - {rec['max_sec']:.0f}s", file=sys.stderr)
+            print(f"    Under 60s:    {rec['pct_under_60s']:.0f}%", file=sys.stderr)
+            print(f"    Under 5min:   {rec['pct_under_300s']:.0f}%", file=sys.stderr)
+        print(f"\n  Stats written to: {STATS_FILE}", file=sys.stderr)
+        print(f"{'='*60}", file=sys.stderr)
+
+
+# ── Main ─────────────────────────────────────────────────────────────────────
+
+def normalize(v):
+    n = np.linalg.norm(v)
+    return v / n if n > 0 else v
+
+
+def main():
+    parser = argparse.ArgumentParser(description="Personalized Bluesky feed")
+    parser.add_argument("token", help="Divepool bearer token")
+    parser.add_argument("did", help="Your AT Protocol DID")
+    parser.add_argument("--window", type=int, required=True,
+                        help="Selection window in seconds")
+    parser.add_argument("--top", type=int, default=3,
+                        help="Max posts to show per window (default: 3)")
+    parser.add_argument("--floor", type=float, default=0.5,
+                        help="Minimum composite score to consider (default: 0.5)")
+    args = parser.parse_args()
+
+    token, user_did = args.token, args.did
+    floor = args.floor
+
+    # ── Step 1: Get your post clusters + medoid embeddings ───────────────
+    print("\nProfiling your posts...", file=sys.stderr, flush=True)
+
+    medoids = []
+    try:
+        data = divepool_search(token, "", limit=1000, did=user_did, cluster=True)
+        for cl in data.get("clusters", []):
+            emb = cl.get("medoid_embedding", [])
+            topics = cl.get("topics", [])
+            if emb and topics and cl.get("size", 0) >= 2:
+                label = " ".join(topics[:3])
+                medoids.append((label, normalize(np.array(emb, dtype=np.float32))))
+    except Exception as e:
+        print(f"  error: {e}", file=sys.stderr, flush=True)
+
+    print(f"  {len(medoids)} clusters from your posts:", file=sys.stderr, flush=True)
+    for label, _ in medoids:
+        print(f"    - {label}", file=sys.stderr, flush=True)
+
+    if not medoids:
+        print("No interest clusters found.", file=sys.stderr)
+        return
+
+    medoid_labels = [m[0] for m in medoids]
+    medoid_matrix = np.stack([m[1] for m in medoids])  # (N, dim)
+    print(f"\n{len(medoids)} medoids, window={args.window}s, top={args.top}, floor={floor}",
+          file=sys.stderr, flush=True)
+
+    # ── Step 2: Stream firehose, score every post, pick best per window ──
+    print("Starting live feed... (Ctrl-C to stop)\n", file=sys.stderr, flush=True)
+
+    firehose_buffer = []
+    stop_event = threading.Event()
+    stream_thread = threading.Thread(
+        target=stream_firehose,
+        args=(token, firehose_buffer, stop_event),
+        daemon=True,
+    )
+    stream_thread.start()
+
+    seen_rkeys = set()
+    did_freq = defaultdict(int)
+    stats = FilterStats()
+    shown_scores = []
+
+    # Window state: candidates collected during current window
+    window_start = time.time()
+    # Each candidate: (composite_score, best_sim, best_label, other_labels, did, col, rkey)
+    candidates = []
+
+    def is_spam(did, text):
+        if did_freq[did] > 5:
+            return True
+        if len(text) < 20:
+            return True
+        if text.count("#") > 8:
+            return True
+        if text.count("http") > 3:
+            return True
+        return False
+
+    def show(handle, text, rkey, comp_score, best_sim, label):
+        link = bsky_link(handle, rkey)
+        gold = ""
+        if len(shown_scores) >= 10:
+            mean = np.mean(shown_scores)
+            std = np.std(shown_scores)
+            if std > 0 and comp_score >= mean + 1.5 * std:
+                gold = " *"
+        shown_scores.append(comp_score)
+        print(f"[{comp_score:.2f}{gold} {label}]\n{text}\n{link} — @{handle}\n", flush=True)
+        stats.record_shown(best_sim)
+
+    def flush_window():
+        """Pick the best candidates from the window, resolve and show them."""
+        nonlocal window_start, candidates
+        stats.windows_elapsed += 1
+
+        if not candidates:
+            stats.windows_empty += 1
+            window_start = time.time()
+            candidates = []
+            return
+
+        # Sort by composite score, pick top N
+        candidates.sort(key=lambda c: -c[0])
+        picks = candidates[:args.top]
+
+        shown_in_window = 0
+        for comp_score, best_sim, best_label, other_labels, did, col, rkey in picks:
+            handle, text = resolve_post_text(did, col, rkey)
+            if not handle or not text:
+                stats.fail_resolve += 1
+                continue
+            if is_spam(did, text):
+                stats.fail_spam += 1
+                continue
+            seen_rkeys.add(rkey)
+            if other_labels:
+                label = best_label + " + " + " + ".join(other_labels[:2])
+            else:
+                label = best_label
+            stats.medoid_hits[best_label] += 1
+            show(handle, text, rkey, comp_score, best_sim, label)
+            shown_in_window += 1
+
+        if shown_in_window == 0:
+            stats.windows_empty += 1
+
+        window_start = time.time()
+        candidates = []
+        did_freq.clear()
+
+    last_stats = time.time()
+
+    try:
+        while True:
+            # Drain firehose buffer, score everything
+            while firehose_buffer:
+                did, col, rkey, lang, c_emb = firehose_buffer.pop(0)
+                stats.total += 1
+                did_freq[did] += 1
+                stats.record_age(rkey)
+
+                if rkey in seen_rkeys or did == user_did:
+                    stats.skip_seen_self += 1
+                    continue
+
+                emb = normalize(np.array(c_emb, dtype=np.float32))
+                sims = medoid_matrix @ emb  # (N,) dot products
+                best_idx = int(np.argmax(sims))
+                best_sim = float(sims[best_idx])
+
+                # Composite: best similarity + small bonus for multi-interest hits
+                # Multi-hit bar: must be genuinely similar, not just baseline noise
+                multi_bar = max(0.7, best_sim * 0.85)
+                multi_hits = [i for i in range(len(sims))
+                              if i != best_idx and sims[i] >= multi_bar]
+                comp_score = best_sim + min(0.15, 0.05 * len(multi_hits))
+                stats.record_score(best_sim)  # histogram tracks raw sim
+
+                if comp_score < floor:
+                    continue
+
+                other_labels = [medoid_labels[i] for i in multi_hits[:2]]
+                candidates.append((
+                    comp_score, best_sim, medoid_labels[best_idx],
+                    other_labels, did, col, rkey,
+                ))
+                # Keep candidate list bounded — only need top picks + some margin
+                # for resolve/spam failures. Sort and trim when it gets large.
+                max_keep = args.top * 5
+                if len(candidates) > max_keep * 2:
+                    candidates.sort(key=lambda c: -c[0])
+                    candidates = candidates[:max_keep]
+
+            # Check if window is up
+            now = time.time()
+            if now - window_start >= args.window:
+                flush_window()
+
+            # Write stats every 30s
+            if now - last_stats >= 30:
+                rate = stats.shown_per_min()
+                age = stats.age_stats()
+                age_str = f" age_med={age['median_sec']:.0f}s" if age else ""
+                n_cand = len(candidates)
+                best_cand = f" best={max(c[0] for c in candidates):.2f}" if candidates else ""
+                print(
+                    f"[stats] scanned={stats.total} shown={stats.shown} "
+                    f"rate={rate:.1f}/min cands={n_cand}{best_cand}{age_str}",
+                    file=sys.stderr, flush=True,
+                )
+                stats.write_file(floor)
+                last_stats = now
+
+            time.sleep(0.05)
+
+    except KeyboardInterrupt:
+        # Flush any remaining candidates
+        if candidates:
+            flush_window()
+        stats.write_file(floor)
+        stats.print_summary(floor)
+        stop_event.set()
+
+
+if __name__ == "__main__":
+    main()
diff --git a/experiments/for_you_v2.py b/experiments/for_you_v2.py
new file mode 100644
index 0000000..02ec58b
--- /dev/null
+++ b/experiments/for_you_v2.py
@@ -0,0 +1,548 @@
+#!/usr/bin/env python3
+"""
+For You v2 — personalized Bluesky feed from the Divepool firehose.
+Improves on for_you.py: per-post embeddings instead of blurry medoid centroids,
+IDF-style weighting to suppress generic interests.
+
+Up to 1000 actual post embeddings as reference vectors. Specificity weighting
+ensures distinctive interests score higher than catch-all clusters.
+Works with or without bearer token (128d vs 768d embeddings).
+
+Usage:
+    python3 for_you_v2.py [token] <did> --window 60 [--top 3]
+"""
+
+import argparse
+import json
+import os
+import sys
+import threading
+import time
+import urllib.error
+import urllib.parse
+import urllib.request
+from collections import defaultdict
+
+import numpy as np
+
+DIVEPOOL_SEARCH = "https://divepool.social/api/v1/search"
+DIVEPOOL_STREAM = "https://divepool.social/api/v1/embeddings"
+BSKY_API = "https://public.api.bsky.app/xrpc"
+STATS_FILE = os.path.join(os.path.dirname(os.path.abspath(__file__)), "for_you_v2_stats.json")
+
+
+# ── API helpers ──────────────────────────────────────────────────────────────
+
+def divepool_search(token, query, limit=100, did=None, cluster=False,
+                    distinct=True, include_embeddings=False):
+    payload = {"query": query, "limit": limit, "distinct": distinct}
+    if did:
+        payload["did"] = did
+    if cluster:
+        payload["cluster"] = True
+    if include_embeddings:
+        payload["include_embeddings"] = True
+    data = json.dumps(payload).encode()
+    headers = {"Content-Type": "application/json"}
+    if token:
+        headers["Authorization"] = f"Bearer {token}"
+    req = urllib.request.Request(DIVEPOOL_SEARCH, data=data, headers=headers,
+                                method="POST")
+    with urllib.request.urlopen(req, timeout=60) as resp:
+        return json.loads(resp.read())
+
+
+def resolve_post_text(did, collection, rkey):
+    uri = f"at://{did}/{collection}/{rkey}"
+    url = f"{BSKY_API}/app.bsky.feed.getPostThread?uri={urllib.parse.quote(uri)}&depth=0"
+    req = urllib.request.Request(url, headers={"Accept": "application/json"})
+    try:
+        with urllib.request.urlopen(req, timeout=5) as resp:
+            data = json.loads(resp.read())
+            post = data.get("thread", {}).get("post", {})
+            text = post.get("record", {}).get("text", "")
+            handle = post.get("author", {}).get("handle", "")
+            return handle, text
+    except Exception:
+        return None, None
+
+
+def bsky_link(handle, rkey):
+    return f"https://bsky.app/profile/{handle}/post/{rkey}"
+
+
+def tid_to_timestamp(tid):
+    """Decode an AT Protocol TID (base32-sortable) to a Unix timestamp in seconds."""
+    charset = "234567abcdefghijklmnopqrstuvwxyz"
+    try:
+        n = 0
+        for ch in tid:
+            n = n * 32 + charset.index(ch)
+        usec = n >> 10
+        return usec / 1_000_000
+    except (ValueError, IndexError):
+        return 0.0
+
+
+# ── Firehose streaming ──────────────────────────────────────────────────────
+
+def stream_firehose(token, post_queue, stop_event):
+    import zstandard as zstd
+    while not stop_event.is_set():
+        try:
+            headers = {}
+            if token:
+                headers["Authorization"] = f"Bearer {token}"
+            req = urllib.request.Request(DIVEPOOL_STREAM, headers=headers)
+            with urllib.request.urlopen(req, timeout=30) as resp:
+                dctx = zstd.ZstdDecompressor()
+                reader = dctx.stream_reader(resp)
+                buf = b""
+                while not stop_event.is_set():
+                    chunk = reader.read(8192)
+                    if not chunk:
+                        break
+                    buf += chunk
+                    while b"\n" in buf:
+                        line, buf = buf.split(b"\n", 1)
+                        if not line.strip():
+                            continue
+                        try:
+                            batch = json.loads(line)
+                        except json.JSONDecodeError:
+                            continue
+                        if not isinstance(batch, dict):
+                            continue
+                        dids = batch.get("did")
+                        if not isinstance(dids, list) or not dids:
+                            continue
+                        cols = batch.get("col", [])
+                        rkeys = batch.get("rkey", [])
+                        langs = batch.get("lang", [])
+                        cs = batch.get("c", [])
+                        for i in range(len(dids)):
+                            col = cols[i] if i < len(cols) else ""
+                            if "feed.post" not in col:
+                                continue
+                            c_emb = cs[i] if i < len(cs) else []
+                            if not c_emb:
+                                continue
+                            rkey = rkeys[i] if i < len(rkeys) else ""
+                            lang = langs[i] if i < len(langs) else ""
+                            post_queue.append((dids[i], col, rkey, lang, c_emb))
+        except Exception:
+            if not stop_event.is_set():
+                time.sleep(2)
+
+
+# ── Stats tracker ──────────────────────────────────────────────────────────
+
+class FilterStats:
+    """Tracks scoring distribution and selection stats."""
+
+    BINS = [0.0, 0.3, 0.4, 0.5, 0.55, 0.6, 0.65, 0.7, 0.75, 0.8, 0.85, 0.9, 0.95, 1.0]
+
+    def __init__(self):
+        self.started = time.time()
+        self.total = 0
+        self.skip_seen_self = 0
+        self.fail_resolve = 0
+        self.fail_spam = 0
+        self.shown = 0
+        self.windows_elapsed = 0
+        self.windows_empty = 0
+        self.score_hist = np.zeros(len(self.BINS) - 1, dtype=np.int64)
+        self.shown_hist = np.zeros(len(self.BINS) - 1, dtype=np.int64)
+        self.cluster_hits = defaultdict(int)
+        self._shown_times = []
+        self._recent_ages = []
+
+    def record_age(self, rkey):
+        ts = tid_to_timestamp(rkey)
+        if ts > 0:
+            age = time.time() - ts
+            self._recent_ages.append((time.time(), age))
+
+    def age_stats(self):
+        now = time.time()
+        cutoff = now - 300
+        self._recent_ages = [(t, a) for t, a in self._recent_ages if t > cutoff]
+        if not self._recent_ages:
+            return None
+        ages = [a for _, a in self._recent_ages]
+        return {
+            "count": len(ages),
+            "median_sec": round(float(np.median(ages)), 1),
+            "p90_sec": round(float(np.percentile(ages, 90)), 1),
+            "max_sec": round(max(ages), 1),
+            "min_sec": round(min(ages), 1),
+            "pct_under_60s": round(100 * sum(1 for a in ages if a < 60) / len(ages), 1),
+            "pct_under_300s": round(100 * sum(1 for a in ages if a < 300) / len(ages), 1),
+        }
+
+    def record_score(self, score):
+        idx = np.searchsorted(self.BINS, score, side="right") - 1
+        idx = max(0, min(idx, len(self.score_hist) - 1))
+        self.score_hist[idx] += 1
+
+    def record_shown(self, score):
+        self.shown += 1
+        self._shown_times.append(time.time())
+        idx = np.searchsorted(self.BINS, score, side="right") - 1
+        idx = max(0, min(idx, len(self.shown_hist) - 1))
+        self.shown_hist[idx] += 1
+
+    def shown_per_min(self):
+        now = time.time()
+        cutoff = now - 300
+        self._shown_times = [t for t in self._shown_times if t > cutoff]
+        if not self._shown_times:
+            return 0.0
+        window = now - self._shown_times[0]
+        if window < 10:
+            return 0.0
+        return len(self._shown_times) / (window / 60)
+
+    def _hist_to_dict(self, hist):
+        out = {}
+        for i, count in enumerate(hist):
+            if count > 0:
+                lo, hi = self.BINS[i], self.BINS[i + 1]
+                out[f"{lo:.2f}-{hi:.2f}"] = int(count)
+        return out
+
+    def to_dict(self):
+        elapsed = time.time() - self.started
+        scored = self.total - self.skip_seen_self
+        return {
+            "elapsed_min": round(elapsed / 60, 1),
+            "shown_per_min": round(self.shown_per_min(), 2),
+            "counts": {
+                "total": self.total,
+                "skip_seen_self": self.skip_seen_self,
+                "scored": scored,
+                "fail_resolve": self.fail_resolve,
+                "fail_spam": self.fail_spam,
+                "shown": self.shown,
+                "windows": self.windows_elapsed,
+                "windows_empty": self.windows_empty,
+            },
+            "rates": {
+                "shown_pct": round(100 * self.shown / scored, 4) if scored else 0,
+                "window_hit_pct": round(100 * (self.windows_elapsed - self.windows_empty)
+                                        / self.windows_elapsed, 1) if self.windows_elapsed else 0,
+            },
+            "score_histogram": self._hist_to_dict(self.score_hist),
+            "shown_histogram": self._hist_to_dict(self.shown_hist),
+            "cluster_hits": dict(sorted(self.cluster_hits.items(), key=lambda x: -x[1])),
+            "recency": self.age_stats(),
+        }
+
+    def write_file(self):
+        data = self.to_dict()
+        tmp = STATS_FILE + ".tmp"
+        with open(tmp, "w") as f:
+            json.dump(data, f, indent=2)
+            f.write("\n")
+        os.replace(tmp, STATS_FILE)
+
+    def print_summary(self):
+        d = self.to_dict()
+        c = d["counts"]
+        r = d["rates"]
+        print(f"\n{'='*60}", file=sys.stderr)
+        print(f"Filter stats  ({d['elapsed_min']} min)",
+              file=sys.stderr)
+        print(f"{'='*60}", file=sys.stderr)
+        print(f"  Total scanned:     {c['total']:>8}", file=sys.stderr)
+        print(f"  Skip seen/self:    {c['skip_seen_self']:>8}", file=sys.stderr)
+        print(f"  Scored:            {c['scored']:>8}", file=sys.stderr)
+        print(f"  Fail resolve:      {c['fail_resolve']:>8}", file=sys.stderr)
+        print(f"  Fail spam:         {c['fail_spam']:>8}", file=sys.stderr)
+        print(f"  Shown:             {c['shown']:>8}  "
+              f"({r['shown_pct']:.3f}% of scored)", file=sys.stderr)
+        print(f"  Windows:           {c['windows']:>8}  "
+              f"({c['windows_empty']} empty, {r['window_hit_pct']:.0f}% hit)",
+              file=sys.stderr)
+        print(f"\n  Shown/min (5m window): {d['shown_per_min']}", file=sys.stderr)
+        def print_hist(title, hist):
+            if not hist:
+                return
+            peak = max(hist.values())
+            print(f"\n  {title}:", file=sys.stderr)
+            for bucket, count in hist.items():
+                bar = "#" * max(1, round(50 * count / peak)) if count else ""
+                print(f"    {bucket}: {count:>7}  {bar}", file=sys.stderr)
+
+        print_hist("Score distribution (all scored posts)", d["score_histogram"])
+        print_hist("Shown post scores", d["shown_histogram"])
+        if d["cluster_hits"]:
+            print(f"\n  Cluster hits (shown posts):", file=sys.stderr)
+            for label, count in d["cluster_hits"].items():
+                print(f"    {count:>6}  {label}", file=sys.stderr)
+        rec = d.get("recency")
+        if rec:
+            print(f"\n  Post recency (5m window, {rec['count']} posts):", file=sys.stderr)
+            print(f"    Median age:   {rec['median_sec']:.0f}s", file=sys.stderr)
+            print(f"    P90 age:      {rec['p90_sec']:.0f}s", file=sys.stderr)
+            print(f"    Range:        {rec['min_sec']:.0f}s - {rec['max_sec']:.0f}s", file=sys.stderr)
+            print(f"    Under 60s:    {rec['pct_under_60s']:.0f}%", file=sys.stderr)
+            print(f"    Under 5min:   {rec['pct_under_300s']:.0f}%", file=sys.stderr)
+        print(f"\n  Stats written to: {STATS_FILE}", file=sys.stderr)
+        print(f"{'='*60}", file=sys.stderr)
+
+
+# ── Main ─────────────────────────────────────────────────────────────────────
+
+def normalize(v):
+    n = np.linalg.norm(v)
+    return v / n if n > 0 else v
+
+
+def main():
+    parser = argparse.ArgumentParser(description="Personalized Bluesky feed (v2 — per-post embeddings)")
+    parser.add_argument("token", nargs="?", default="",
+                        help="Divepool bearer token (optional — without token, 128d embeddings)")
+    parser.add_argument("did", help="Your AT Protocol DID")
+    parser.add_argument("--window", type=int, required=True,
+                        help="Selection window in seconds")
+    parser.add_argument("--top", type=int, default=3,
+                        help="Max posts to show per window (default: 3)")
+    args = parser.parse_args()
+
+    token, user_did = args.token, args.did
+
+    # ── Step 1: Fetch your posts with per-result embeddings ─────────────
+    print("\nProfiling your posts...", file=sys.stderr, flush=True)
+
+    cluster_id_to_label = {}
+    ref_labels = []
+    ref_cluster_ids = []
+    ref_vecs = []
+
+    try:
+        data = divepool_search(token, "", limit=1000, did=user_did,
+                               cluster=True, include_embeddings=True)
+
+        # Build cluster label lookup
+        for cl in data.get("clusters", []):
+            topics = cl.get("topics", [])
+            if topics:
+                cluster_id_to_label[cl["id"]] = " ".join(topics[:3])
+
+        # Extract per-result embeddings
+        for r in data.get("results", []):
+            emb = r.get("embedding", [])
+            if not emb:
+                continue
+            cid = r.get("cluster_id")
+            if cid is not None and cid in cluster_id_to_label:
+                label = cluster_id_to_label[cid]
+            else:
+                # Unclustered result — use per-result topics or fallback
+                topics = r.get("topics", [])
+                label = " ".join(topics[:3]) if topics else "unclustered"
+                if cid is not None:
+                    cluster_id_to_label[cid] = label
+
+            ref_labels.append(label)
+            ref_cluster_ids.append(cid if cid is not None else -1)
+            ref_vecs.append(normalize(np.array(emb, dtype=np.float32)))
+
+    except Exception as e:
+        print(f"  error: {e}", file=sys.stderr, flush=True)
+
+    if not ref_vecs:
+        print("No reference embeddings found.", file=sys.stderr)
+        return
+
+    ref_matrix = np.stack(ref_vecs)  # (N, dim)
+    n_clusters = len(set(ref_cluster_ids) - {-1})
+
+    print(f"  {len(ref_vecs)} reference embeddings from {n_clusters} clusters:",
+          file=sys.stderr, flush=True)
+    for cid, label in sorted(cluster_id_to_label.items()):
+        count = ref_cluster_ids.count(cid)
+        print(f"    - {label} ({count} posts)", file=sys.stderr, flush=True)
+
+    # ── Step 2: IDF-style weighting — downweight generic references ─────
+    print("  Computing specificity weights...", file=sys.stderr, flush=True)
+    self_sims = ref_matrix @ ref_matrix.T  # (N, N)
+    np.fill_diagonal(self_sims, 0)
+    mean_sims = self_sims.mean(axis=1)  # how similar each ref is to all others
+    ref_weights = 1.0 - mean_sims       # distinctive refs get higher weight
+    ref_weights = ref_weights / ref_weights.max()  # normalize: most distinctive = 1.0
+    ref_weights = np.clip(ref_weights, 0.3, 1.0)   # floor at 0.3 so nothing is crushed
+
+    # Show weight distribution per cluster
+    for cid, label in sorted(cluster_id_to_label.items()):
+        mask = [i for i, c in enumerate(ref_cluster_ids) if c == cid]
+        if mask:
+            w = ref_weights[mask]
+            print(f"      {label}: weight {w.mean():.2f} (min={w.min():.2f}, max={w.max():.2f})",
+                  file=sys.stderr, flush=True)
+
+    print(f"\n{len(ref_vecs)} refs, {n_clusters} clusters, "
+          f"window={args.window}s, top={args.top}",
+          file=sys.stderr, flush=True)
+
+    # ── Step 3: Stream firehose, score every post, pick best per window ──
+    print("Starting live feed... (Ctrl-C to stop)\n", file=sys.stderr, flush=True)
+
+    firehose_buffer = []
+    stop_event = threading.Event()
+    stream_thread = threading.Thread(
+        target=stream_firehose,
+        args=(token, firehose_buffer, stop_event),
+        daemon=True,
+    )
+    stream_thread.start()
+
+    seen_rkeys = set()
+    did_freq = defaultdict(int)
+    stats = FilterStats()
+    shown_scores = []
+
+    window_start = time.time()
+    # Each candidate: (comp_score, best_sim, best_label, other_labels, did, col, rkey)
+    candidates = []
+
+    def is_spam(did, text):
+        if did_freq[did] > 5:
+            return True
+        if len(text) < 20:
+            return True
+        if text.count("#") > 8:
+            return True
+        if text.count("http") > 3:
+            return True
+        return False
+
+    def show(handle, text, rkey, comp_score, best_sim, label):
+        link = bsky_link(handle, rkey)
+        gold = ""
+        if len(shown_scores) >= 10:
+            mean = np.mean(shown_scores)
+            std = np.std(shown_scores)
+            if std > 0 and comp_score >= mean + 1.5 * std:
+                gold = " *"
+        shown_scores.append(comp_score)
+        print(f"[{comp_score:.2f}{gold} {label}]\n{text}\n{link} — @{handle}\n", flush=True)
+        stats.record_shown(best_sim)
+
+    def flush_window():
+        """Pick the best candidates from the window, resolve and show them."""
+        nonlocal window_start, candidates
+        stats.windows_elapsed += 1
+
+        if not candidates:
+            stats.windows_empty += 1
+            window_start = time.time()
+            candidates = []
+            return
+
+        candidates.sort(key=lambda c: -c[0])
+        picks = candidates[:args.top]
+
+        shown_in_window = 0
+        for comp_score, best_sim, best_label, other_labels, did, col, rkey in picks:
+            handle, text = resolve_post_text(did, col, rkey)
+            if not handle or not text:
+                stats.fail_resolve += 1
+                continue
+            if is_spam(did, text):
+                stats.fail_spam += 1
+                continue
+            seen_rkeys.add(rkey)
+            if other_labels:
+                label = best_label + " + " + " + ".join(other_labels[:2])
+            else:
+                label = best_label
+            stats.cluster_hits[best_label] += 1
+            show(handle, text, rkey, comp_score, best_sim, label)
+            shown_in_window += 1
+
+        if shown_in_window == 0:
+            stats.windows_empty += 1
+
+        window_start = time.time()
+        candidates = []
+        did_freq.clear()
+
+    last_stats = time.time()
+
+    try:
+        while True:
+            # Drain firehose buffer, score everything
+            while firehose_buffer:
+                did, col, rkey, lang, c_emb = firehose_buffer.pop(0)
+                stats.total += 1
+                did_freq[did] += 1
+                stats.record_age(rkey)
+
+                if rkey in seen_rkeys or did == user_did:
+                    stats.skip_seen_self += 1
+                    continue
+
+                emb = normalize(np.array(c_emb, dtype=np.float32))
+                raw_sims = ref_matrix @ emb           # (N,)
+                weighted_sims = raw_sims * ref_weights # IDF-weighted
+
+                best_idx = int(np.argmax(weighted_sims))
+                best_sim = float(weighted_sims[best_idx])
+
+                # Multi-interest: count distinct clusters with high hits
+                multi_bar = max(0.7, best_sim * 0.85)
+                best_cluster = ref_cluster_ids[best_idx]
+                hit_clusters = set()
+                for i in range(len(weighted_sims)):
+                    if weighted_sims[i] >= multi_bar and ref_cluster_ids[i] != best_cluster:
+                        hit_clusters.add(ref_cluster_ids[i])
+                comp_score = best_sim + min(0.15, 0.05 * len(hit_clusters))
+                stats.record_score(best_sim)
+
+                best_label = ref_labels[best_idx]
+                other_labels = [cluster_id_to_label.get(cid, "?")
+                                for cid in list(hit_clusters)[:2]]
+                candidates.append((
+                    comp_score, best_sim, best_label,
+                    other_labels, did, col, rkey,
+                ))
+                # Keep candidate list bounded
+                max_keep = args.top * 5
+                if len(candidates) > max_keep * 2:
+                    candidates.sort(key=lambda c: -c[0])
+                    candidates = candidates[:max_keep]
+
+            # Check if window is up
+            now = time.time()
+            if now - window_start >= args.window:
+                flush_window()
+
+            # Write stats every 30s
+            if now - last_stats >= 30:
+                rate = stats.shown_per_min()
+                age = stats.age_stats()
+                age_str = f" age_med={age['median_sec']:.0f}s" if age else ""
+                n_cand = len(candidates)
+                best_cand = f" best={max(c[0] for c in candidates):.2f}" if candidates else ""
+                print(
+                    f"[stats] scanned={stats.total} shown={stats.shown} "
+                    f"rate={rate:.1f}/min cands={n_cand}{best_cand}{age_str}",
+                    file=sys.stderr, flush=True,
+                )
+                stats.write_file()
+                last_stats = now
+
+            time.sleep(0.05)
+
+    except KeyboardInterrupt:
+        if candidates:
+            flush_window()
+        stats.write_file()
+        stats.print_summary()
+        stop_event.set()
+
+
+if __name__ == "__main__":
+    main()
diff --git a/experiments/live_trends.py b/experiments/live_trends.py
new file mode 100644
index 0000000..b6787b3
--- /dev/null
+++ b/experiments/live_trends.py
@@ -0,0 +1,304 @@
+#!/usr/bin/env python3
+"""
+Live Trends — taps the Divepool embedding firehose for a window,
+clusters the incoming embeddings in real time, then uses the search API
+to label each cluster with human-readable topics.
+
+Requires: numpy, scikit-learn (for local HDBSCAN/UMAP-lite clustering)
+Falls back to simpler k-means if HDBSCAN unavailable.
+
+Usage:
+    python3 live_trends.py <bearer_token> [--seconds 30] [--top 5]
+"""
+
+import argparse
+import json
+import math
+import struct
+import sys
+import time
+import urllib.request
+from collections import Counter, defaultdict
+
+import numpy as np
+
+DIVEPOOL_STREAM = "https://divepool.social/api/v1/embeddings"
+DIVEPOOL_SEARCH = "https://divepool.social/api/v1/search"
+
+
+def stream_embeddings(token: str, seconds: int) -> list[dict]:
+    """Collect embeddings from the firehose for `seconds` seconds."""
+    req = urllib.request.Request(DIVEPOOL_STREAM, headers={
+        "Authorization": f"Bearer {token}",
+    })
+    events = []
+    print(f"  Streaming for {seconds}s...", end="", flush=True)
+    t0 = time.time()
+
+    try:
+        with urllib.request.urlopen(req, timeout=seconds + 10) as resp:
+            # The stream is zstd-compressed NDJSON — we need to decompress.
+            # Use zstandard if available, else fall back to subprocess.
+            try:
+                import zstandard as zstd
+                dctx = zstd.ZstdDecompressor()
+                reader = dctx.stream_reader(resp)
+            except ImportError:
+                # Fall back to piping through zstd CLI
+                import subprocess, io
+                proc = subprocess.Popen(
+                    ["zstd", "-d"],
+                    stdin=subprocess.PIPE,
+                    stdout=subprocess.PIPE,
+                    stderr=subprocess.DEVNULL,
+                )
+                # We need to pump data from resp to proc.stdin in a thread
+                import threading
+                def pump():
+                    try:
+                        while True:
+                            chunk = resp.read(8192)
+                            if not chunk:
+                                break
+                            proc.stdin.write(chunk)
+                    except Exception:
+                        pass
+                    finally:
+                        proc.stdin.close()
+                t = threading.Thread(target=pump, daemon=True)
+                t.start()
+                reader = proc.stdout
+
+            buf = b""
+            while time.time() - t0 < seconds:
+                chunk = reader.read(4096)
+                if not chunk:
+                    break
+                buf += chunk
+                while b"\n" in buf:
+                    line, buf = buf.split(b"\n", 1)
+                    if not line.strip():
+                        continue
+                    try:
+                        batch = json.loads(line)
+                    except json.JSONDecodeError:
+                        continue
+                    # Batch is columnar: parallel arrays keyed by "did", "col", etc.
+                    # Skip heartbeats (empty batches) or malformed data.
+                    if not isinstance(batch, dict):
+                        continue
+                    dids = batch.get("did")
+                    if not isinstance(dids, list) or len(dids) == 0:
+                        continue
+                    cols = batch.get("col", [])
+                    rkeys = batch.get("rkey", [])
+                    langs = batch.get("lang", [])
+                    cs = batch.get("c", [])
+                    rs = batch.get("r", [])
+                    for i in range(len(dids)):
+                        c_emb = cs[i] if i < len(cs) else []
+                        r_emb = rs[i] if i < len(rs) else []
+                        col = cols[i] if i < len(cols) else ""
+                        lang = langs[i] if i < len(langs) else ""
+                        rkey = rkeys[i] if i < len(rkeys) else ""
+                        if c_emb:
+                            events.append({
+                                "did": dids[i],
+                                "col": col,
+                                "rkey": rkey,
+                                "lang": lang,
+                                "c": c_emb,
+                                "r": r_emb,
+                            })
+                    if len(events) % 100 == 0:
+                        print(f"\r  Streaming for {seconds}s... {len(events)} events", end="", flush=True)
+    except Exception as e:
+        print(f"\n  Stream ended: {e}")
+
+    elapsed = time.time() - t0
+    print(f"\r  Collected {len(events)} events in {elapsed:.1f}s ({len(events)/max(elapsed,1):.0f}/s)")
+    return events
+
+
+def cluster_embeddings(events: list[dict], n_clusters: int = 8) -> list[dict]:
+    """Cluster events by their C (cluster) embeddings using k-means."""
+    if len(events) < n_clusters * 2:
+        n_clusters = max(2, len(events) // 5)
+
+    from sklearn.cluster import MiniBatchKMeans
+    from sklearn.preprocessing import normalize
+
+    # Build embedding matrix
+    C = np.array([e["c"] for e in events], dtype=np.float32)
+    C = normalize(C)  # L2 normalize
+
+    km = MiniBatchKMeans(n_clusters=n_clusters, random_state=42, n_init=3)
+    labels = km.fit_predict(C)
+
+    # Group events by cluster
+    clusters = defaultdict(list)
+    for i, label in enumerate(labels):
+        clusters[label].append(events[i])
+
+    # Sort clusters by size
+    result = []
+    for label, members in sorted(clusters.items(), key=lambda x: -len(x[1])):
+        # Compute centroid
+        centroid = np.mean([m["c"] for m in members], axis=0)
+        result.append({
+            "id": label,
+            "size": len(members),
+            "members": members,
+            "centroid": centroid.tolist(),
+            "langs": Counter(m["lang"] for m in members).most_common(3),
+            "collections": Counter(m["col"] for m in members).most_common(3),
+        })
+    return result
+
+
+def label_cluster_via_search(token: str, cluster: dict) -> dict | None:
+    """Use a representative member's embedding to find similar indexed posts."""
+    # Pick a few members and search for their content to get labels
+    members = cluster["members"]
+    # Use a random sample member's AT URI to find its text via search
+    # Instead, we'll search with the cluster centroid's nearest indexed neighbor
+    # by searching for posts from the same DIDs
+    sample = members[:3]
+    sample_dids = [m["did"] for m in sample]
+
+    # We can't search by embedding directly, but we can search for content
+    # that's semantically similar. Let's use the collection type as a hint.
+    col_type = cluster["collections"][0][0] if cluster["collections"] else ""
+
+    # Search for posts by these authors — the search API takes text queries,
+    # so we'll use a representative DID handle lookup
+    # Actually, let's just use the search API with clustering to find what topics
+    # match this cluster's centroid. We'll pick a member and look up its text.
+    return None  # We'll use a different approach below
+
+
+def search_for_context(token: str, query: str, limit: int = 50) -> dict:
+    payload = json.dumps({
+        "query": query, "limit": limit,
+        "cluster": True, "distinct": True,
+    }).encode()
+    req = urllib.request.Request(DIVEPOOL_SEARCH, data=payload, headers={
+        "Content-Type": "application/json",
+        "Authorization": f"Bearer {token}",
+    }, method="POST")
+    with urllib.request.urlopen(req, timeout=15) as resp:
+        return json.loads(resp.read())
+
+
+def resolve_posts(events: list[dict], max_posts: int = 5) -> list[str]:
+    """Resolve AT URIs to post text via the Bluesky API."""
+    texts = []
+    for e in events[:max_posts]:
+        uri = f"at://{e['did']}/{e['col']}/{e['rkey']}"
+        url = f"https://public.api.bsky.app/xrpc/app.bsky.feed.getPostThread?uri={urllib.parse.quote(uri)}&depth=0"
+        req = urllib.request.Request(url, headers={"Accept": "application/json"})
+        try:
+            with urllib.request.urlopen(req, timeout=5) as resp:
+                data = json.loads(resp.read())
+                post = data.get("thread", {}).get("post", {}).get("record", {})
+                text = post.get("text", "")
+                if text:
+                    texts.append(text)
+        except Exception:
+            pass
+    return texts
+
+
+import urllib.parse
+
+
+def main():
+    parser = argparse.ArgumentParser(description="Live Bluesky trend detector")
+    parser.add_argument("token", help="Divepool bearer token")
+    parser.add_argument("--seconds", type=int, default=30, help="Streaming window (default: 30)")
+    parser.add_argument("--clusters", type=int, default=8, help="Number of clusters (default: 8)")
+    parser.add_argument("--resolve", action="store_true", help="Resolve post text via Bluesky API (slower)")
+    args = parser.parse_args()
+
+    print(f"Live Trends — Bluesky Embedding Firehose")
+    print(f"{'─'*50}")
+
+    # Step 1: Stream
+    print(f"\n[1/3] Tapping firehose...")
+    events = stream_embeddings(args.token, args.seconds)
+    if len(events) < 10:
+        print("Too few events to cluster. Try a longer window.")
+        return
+
+    # Filter to posts only
+    posts = [e for e in events if "feed.post" in e.get("col", "")]
+    profiles = [e for e in events if "actor.profile" in e.get("col", "")]
+    print(f"      {len(posts)} posts, {len(profiles)} profile updates")
+
+    if len(posts) < 10:
+        print("Too few posts to cluster.")
+        return
+
+    # Step 2: Cluster
+    print(f"\n[2/3] Clustering {len(posts)} post embeddings...")
+    clusters = cluster_embeddings(posts, n_clusters=args.clusters)
+    print(f"      {len(clusters)} clusters formed")
+
+    # Step 3: Label clusters by resolving sample posts
+    print(f"\n[3/3] Resolving cluster content...")
+    for cl in clusters:
+        members = cl["members"]
+        # Resolve a few posts from each cluster to understand content
+        if args.resolve:
+            texts = resolve_posts(members, max_posts=5)
+        else:
+            texts = []
+        cl["sample_texts"] = texts
+
+        # Language breakdown
+        lang_str = ", ".join(f"{lang}({n})" for lang, n in cl["langs"])
+        cl["lang_str"] = lang_str
+
+        # Unique authors
+        cl["unique_authors"] = len(set(m["did"] for m in members))
+
+    # ── Report ──
+    print(f"\n\n{'━'*50}")
+    print(f"  LIVE TRENDS — {len(posts)} posts in {args.seconds}s")
+    print(f"  ({len(posts)/args.seconds:.1f} posts/sec)")
+    print(f"{'━'*50}")
+
+    # Overall language distribution
+    all_langs = Counter(e["lang"] for e in posts)
+    print(f"\n  Language mix: {', '.join(f'{l}({n})' for l, n in all_langs.most_common(10))}")
+
+    # Collection breakdown
+    all_cols = Counter(e["col"] for e in events)
+    print(f"  Content types: {', '.join(f'{c.split('.')[-1]}({n})' for c, n in all_cols.most_common(5))}")
+
+    for i, cl in enumerate(clusters):
+        pct = cl["size"] / len(posts) * 100
+        print(f"\n  ── Cluster #{cl['id']} — {cl['size']} posts ({pct:.0f}%) — {cl['unique_authors']} authors ──")
+        print(f"     Languages: {cl['lang_str']}")
+
+        if cl["sample_texts"]:
+            print(f"     Sample posts:")
+            for t in cl["sample_texts"][:3]:
+                text = t[:120].replace("\n", " ")
+                print(f"       \"{text}\"")
+        else:
+            # Show AT URIs for manual inspection
+            print(f"     Sample URIs (use --resolve to fetch text):")
+            for m in cl["members"][:3]:
+                print(f"       at://{m['did']}/{m['col']}/{m['rkey']}")
+
+    # Embedding dimensionality check
+    if posts:
+        dim = len(posts[0].get("c", []))
+        print(f"\n  Embedding dimensionality: {dim}d (768d = authenticated)")
+
+    print(f"\n{'━'*50}\n")
+
+
+if __name__ == "__main__":
+    main()
diff --git a/experiments/outliers.py b/experiments/outliers.py
new file mode 100644
index 0000000..8ddb4b7
--- /dev/null
+++ b/experiments/outliers.py
@@ -0,0 +1,366 @@
+#!/usr/bin/env python3
+"""
+Outlier Finder — taps the Divepool firehose, then uses UMAP + HDBSCAN to find
+the surprising, niche, and weird corners of Bluesky.
+
+Unlike k-means, HDBSCAN:
+  - Finds clusters of varying density and size (micro-communities surface)
+  - Labels points that don't fit anywhere as noise (-1) — true outliers
+  - Provides per-point outlier scores for ranking weirdness
+
+We invert the usual presentation: smallest clusters and highest-outlier posts
+come first. The big boring blobs get a one-line summary at the bottom.
+
+Usage:
+    python3 outliers.py <token> [--seconds 600] [--min-cluster 3]
+"""
+
+import argparse
+import json
+import sys
+import time
+import urllib.error
+import urllib.parse
+import urllib.request
+from collections import Counter, defaultdict
+from concurrent.futures import ThreadPoolExecutor, as_completed
+
+import numpy as np
+
+DIVEPOOL_STREAM = "https://divepool.social/api/v1/embeddings"
+BSKY_API = "https://public.api.bsky.app/xrpc"
+
+
+# ── Firehose streaming ──────────────────────────────────────────────────────
+
+def stream_embeddings(token: str, seconds: int) -> list[dict]:
+    req = urllib.request.Request(DIVEPOOL_STREAM, headers={
+        "Authorization": f"Bearer {token}",
+    })
+    events = []
+    print(f"  Streaming for {seconds}s...", end="", flush=True)
+    t0 = time.time()
+
+    try:
+        import zstandard as zstd
+        with urllib.request.urlopen(req, timeout=seconds + 10) as resp:
+            dctx = zstd.ZstdDecompressor()
+            reader = dctx.stream_reader(resp)
+            buf = b""
+            while time.time() - t0 < seconds:
+                chunk = reader.read(8192)
+                if not chunk:
+                    break
+                buf += chunk
+                while b"\n" in buf:
+                    line, buf = buf.split(b"\n", 1)
+                    if not line.strip():
+                        continue
+                    try:
+                        batch = json.loads(line)
+                    except json.JSONDecodeError:
+                        continue
+                    if not isinstance(batch, dict):
+                        continue
+                    dids = batch.get("did")
+                    if not isinstance(dids, list) or len(dids) == 0:
+                        continue
+                    cols = batch.get("col", [])
+                    rkeys = batch.get("rkey", [])
+                    langs = batch.get("lang", [])
+                    cs = batch.get("c", [])
+                    rs = batch.get("r", [])
+                    for i in range(len(dids)):
+                        c_emb = cs[i] if i < len(cs) else []
+                        if not c_emb:
+                            continue
+                        col = cols[i] if i < len(cols) else ""
+                        lang = langs[i] if i < len(langs) else ""
+                        rkey = rkeys[i] if i < len(rkeys) else ""
+                        r_emb = rs[i] if i < len(rs) else []
+                        events.append({
+                            "did": dids[i], "col": col, "rkey": rkey,
+                            "lang": lang, "c": c_emb, "r": r_emb,
+                        })
+                    if len(events) % 200 == 0 and len(events) > 0:
+                        print(f"\r  Streaming for {seconds}s... {len(events)} events", end="", flush=True)
+    except Exception as e:
+        print(f"\n  Stream ended: {e}")
+
+    elapsed = time.time() - t0
+    print(f"\r  Collected {len(events)} events in {elapsed:.1f}s ({len(events)/max(elapsed,1):.0f}/s)")
+    return events
+
+
+# ── Clustering ───────────────────────────────────────────────────────────────
+
+def cluster_with_hdbscan(events: list[dict], min_cluster_size: int = 5):
+    """
+    UMAP (768d → 15d) then HDBSCAN. Returns (labels, outlier_scores, umap_2d).
+    """
+    import umap
+    import hdbscan
+
+    C = np.array([e["c"] for e in events], dtype=np.float32)
+    # L2 normalize
+    norms = np.linalg.norm(C, axis=1, keepdims=True)
+    norms[norms == 0] = 1
+    C = C / norms
+
+    print(f"    UMAP 768d → 15d...", end="", flush=True)
+    t0 = time.time()
+    reducer = umap.UMAP(
+        n_components=15, n_neighbors=30, min_dist=0.0,
+        metric="cosine", random_state=42, low_memory=True,
+    )
+    embedding_15d = reducer.fit_transform(C)
+    print(f" ({time.time()-t0:.1f}s)")
+
+    print(f"    HDBSCAN (min_cluster_size={min_cluster_size})...", end="", flush=True)
+    t0 = time.time()
+    clusterer = hdbscan.HDBSCAN(
+        min_cluster_size=min_cluster_size,
+        min_samples=2,
+        cluster_selection_method="eom",  # excess of mass — favors many small clusters
+        prediction_data=True,
+    )
+    labels = clusterer.fit_predict(embedding_15d)
+    outlier_scores = clusterer.outlier_scores_
+    print(f" ({time.time()-t0:.1f}s)")
+
+    # Also get 2D for potential visualization
+    print(f"    UMAP 15d → 2d...", end="", flush=True)
+    t0 = time.time()
+    reducer_2d = umap.UMAP(
+        n_components=2, n_neighbors=30, min_dist=0.1,
+        metric="euclidean", random_state=42, low_memory=True,
+    )
+    embedding_2d = reducer_2d.fit_transform(embedding_15d)
+    print(f" ({time.time()-t0:.1f}s)")
+
+    return labels, outlier_scores, embedding_2d
+
+
+# ── Post resolution ──────────────────────────────────────────────────────────
+
+def resolve_post(event: dict) -> str | None:
+    uri = f"at://{event['did']}/{event['col']}/{event['rkey']}"
+    url = f"{BSKY_API}/app.bsky.feed.getPostThread?uri={urllib.parse.quote(uri)}&depth=0"
+    req = urllib.request.Request(url, headers={"Accept": "application/json"})
+    try:
+        with urllib.request.urlopen(req, timeout=5) as resp:
+            data = json.loads(resp.read())
+            post = data.get("thread", {}).get("post", {})
+            text = post.get("record", {}).get("text", "")
+            handle = post.get("author", {}).get("handle", "")
+            likes = post.get("likeCount", 0)
+            return f"@{handle}  [{likes}♥]  \"{text}\"" if text else None
+    except Exception:
+        return None
+
+
+def resolve_batch(events: list[dict], max_posts: int = 5) -> list[str]:
+    results = []
+    with ThreadPoolExecutor(max_workers=10) as pool:
+        futures = {pool.submit(resolve_post, e): e for e in events[:max_posts]}
+        for f in as_completed(futures):
+            r = f.result()
+            if r:
+                results.append(r)
+    return results
+
+
+# ── Main ─────────────────────────────────────────────────────────────────────
+
+def main():
+    parser = argparse.ArgumentParser(description="Bluesky outlier/niche finder")
+    parser.add_argument("token", help="Divepool bearer token")
+    parser.add_argument("--seconds", type=int, default=600, help="Streaming window (default: 600)")
+    parser.add_argument("--min-cluster", type=int, default=3, help="HDBSCAN min_cluster_size (default: 3, lower = more micro-clusters)")
+    args = parser.parse_args()
+
+    print(f"Outlier Finder — Bluesky Firehose")
+    print(f"{'─'*55}")
+
+    # ── Stream ───────────────────────────────────────────────────────────
+    print(f"\n[1/4] Tapping firehose...")
+    events = stream_embeddings(args.token, args.seconds)
+
+    posts = [e for e in events if "feed.post" in e.get("col", "")]
+    profiles = [e for e in events if "actor.profile" in e.get("col", "")]
+    print(f"      {len(posts)} posts, {len(profiles)} profile updates")
+
+    if len(posts) < 20:
+        print("Too few posts. Try a longer window.")
+        return
+
+    # ── Cluster ──────────────────────────────────────────────────────────
+    print(f"\n[2/4] UMAP + HDBSCAN clustering...")
+    labels, outlier_scores, coords_2d = cluster_with_hdbscan(posts, args.min_cluster)
+
+    # Attach labels and scores to events
+    for i, e in enumerate(posts):
+        e["cluster"] = int(labels[i])
+        e["outlier_score"] = float(outlier_scores[i])
+        e["x"] = float(coords_2d[i, 0])
+        e["y"] = float(coords_2d[i, 1])
+
+    # Build cluster groups
+    clusters = defaultdict(list)
+    noise = []
+    for e in posts:
+        if e["cluster"] == -1:
+            noise.append(e)
+        else:
+            clusters[e["cluster"]].append(e)
+
+    n_clusters = len(clusters)
+    sizes = sorted([len(m) for m in clusters.values()])
+    print(f"      {n_clusters} clusters, {len(noise)} noise points (true outliers)")
+    if sizes:
+        print(f"      Cluster sizes: min={sizes[0]}, median={sizes[len(sizes)//2]}, max={sizes[-1]}")
+
+    # ── Resolve posts ────────────────────────────────────────────────────
+    # We want to resolve: all small clusters + top outliers + a sample of big clusters.
+    small_clusters = {k: v for k, v in clusters.items() if len(v) <= 15}
+    big_clusters = {k: v for k, v in clusters.items() if len(v) > 15}
+
+    # Count how many posts we need to resolve
+    to_resolve = []
+    for members in small_clusters.values():
+        to_resolve.extend(members[:5])
+    noise_ranked = sorted(noise, key=lambda e: -e["outlier_score"])
+    to_resolve.extend(noise_ranked[:20])
+    for members in big_clusters.values():
+        to_resolve.extend(members[:2])
+
+    # Deduplicate by rkey
+    seen = set()
+    unique_resolve = []
+    for e in to_resolve:
+        if e["rkey"] not in seen:
+            seen.add(e["rkey"])
+            unique_resolve.append(e)
+
+    print(f"\n[3/4] Resolving {len(unique_resolve)} posts via Bluesky API...")
+    t0 = time.time()
+    resolved = {}
+    with ThreadPoolExecutor(max_workers=15) as pool:
+        futures = {pool.submit(resolve_post, e): e for e in unique_resolve}
+        for f in as_completed(futures):
+            e = futures[f]
+            r = f.result()
+            if r:
+                resolved[e["rkey"]] = r
+    print(f"      Resolved {len(resolved)}/{len(unique_resolve)} ({time.time()-t0:.1f}s)")
+
+    # ── Save 2D coordinates for optional visualization ───────────────────
+    print(f"\n[4/4] Saving 2D map to outlier_map.json...")
+    map_data = []
+    for e in posts:
+        map_data.append({
+            "x": e["x"], "y": e["y"],
+            "cluster": e["cluster"],
+            "outlier_score": e["outlier_score"],
+            "did": e["did"], "rkey": e["rkey"],
+            "lang": e["lang"],
+            "text": resolved.get(e["rkey"], ""),
+        })
+    with open("outlier_map.json", "w") as f:
+        json.dump(map_data, f)
+    print(f"      {len(map_data)} points saved")
+
+    # ── REPORT ───────────────────────────────────────────────────────────
+    print(f"\n\n{'━'*55}")
+    print(f"  OUTLIER REPORT — {len(posts)} posts, {n_clusters} clusters, {len(noise)} outliers")
+    print(f"{'━'*55}")
+
+    # ── Section 1: True outliers (noise points, ranked by outlier score)
+    print(f"\n{'─'*55}")
+    print(f"  TRUE OUTLIERS — posts that fit nowhere")
+    print(f"  (HDBSCAN noise points, ranked by outlier score)")
+    print(f"{'─'*55}")
+    shown = 0
+    for e in noise_ranked:
+        text = resolved.get(e["rkey"])
+        if not text:
+            continue
+        # Truncate for display
+        lines = text.split('"')
+        display = text[:200].replace("\n", " ")
+        print(f"\n  [{e['outlier_score']:.3f}] {display}")
+        shown += 1
+        if shown >= 15:
+            break
+
+    # ── Section 2: Micro-clusters (the niche communities)
+    micro = {k: v for k, v in clusters.items() if len(v) <= 10}
+    small = {k: v for k, v in clusters.items() if 10 < len(v) <= 30}
+    big = {k: v for k, v in clusters.items() if len(v) > 30}
+
+    if micro:
+        print(f"\n{'─'*55}")
+        print(f"  MICRO-CLUSTERS — tiny niche communities ({len(micro)} found)")
+        print(f"{'─'*55}")
+        for cid, members in sorted(micro.items(), key=lambda x: len(x[1])):
+            langs = Counter(m["lang"] for m in members).most_common(2)
+            lang_str = ", ".join(f"{l}" for l, _ in langs)
+            n_authors = len(set(m["did"] for m in members))
+            print(f"\n  Cluster #{cid} — {len(members)} posts, {n_authors} authors [{lang_str}]")
+            for m in members[:4]:
+                text = resolved.get(m["rkey"])
+                if text:
+                    display = text[:180].replace("\n", " ")
+                    print(f"    {display}")
+
+    if small:
+        print(f"\n{'─'*55}")
+        print(f"  SMALL CLUSTERS — emerging topics ({len(small)} found)")
+        print(f"{'─'*55}")
+        for cid, members in sorted(small.items(), key=lambda x: len(x[1])):
+            langs = Counter(m["lang"] for m in members).most_common(2)
+            lang_str = ", ".join(f"{l}" for l, _ in langs)
+            n_authors = len(set(m["did"] for m in members))
+            print(f"\n  Cluster #{cid} — {len(members)} posts, {n_authors} authors [{lang_str}]")
+            for m in members[:3]:
+                text = resolved.get(m["rkey"])
+                if text:
+                    display = text[:180].replace("\n", " ")
+                    print(f"    {display}")
+
+    # ── Section 3: Big clusters (one-liner summary)
+    if big:
+        print(f"\n{'─'*55}")
+        print(f"  BIG CLUSTERS — the expected mainstream ({len(big)} found)")
+        print(f"{'─'*55}")
+        for cid, members in sorted(big.items(), key=lambda x: -len(x[1])):
+            pct = len(members) / len(posts) * 100
+            langs = Counter(m["lang"] for m in members).most_common(1)
+            n_authors = len(set(m["did"] for m in members))
+            sample = resolved.get(members[0]["rkey"], "")
+            snippet = sample[:100].replace("\n", " ") if sample else "(unresolved)"
+            print(f"  #{cid:3d}  {len(members):4d} posts ({pct:4.1f}%)  {n_authors:3d} authors  {snippet}")
+
+    # ── Language outliers
+    all_langs = Counter(e["lang"] for e in posts)
+    rare_langs = [(l, n) for l, n in all_langs.most_common() if n <= 5 and l]
+    if rare_langs:
+        print(f"\n{'─'*55}")
+        print(f"  RARE LANGUAGE POSTS")
+        print(f"{'─'*55}")
+        for lang, count in rare_langs:
+            lang_posts = [e for e in posts if e["lang"] == lang]
+            print(f"\n  {lang} ({count} posts):")
+            for e in lang_posts[:3]:
+                text = resolved.get(e["rkey"])
+                if text:
+                    display = text[:180].replace("\n", " ")
+                    print(f"    {display}")
+
+    print(f"\n{'━'*55}")
+    print(f"  2D map saved to outlier_map.json ({len(posts)} points)")
+    print(f"{'━'*55}\n")
+
+
+if __name__ == "__main__":
+    main()
diff --git a/experiments/topic_explorer.py b/experiments/topic_explorer.py
new file mode 100644
index 0000000..af528d5
--- /dev/null
+++ b/experiments/topic_explorer.py
@@ -0,0 +1,288 @@
+#!/usr/bin/env python3
+"""
+Topic Explorer — combines Divepool semantic search with the public Bluesky API
+to map the community structure behind any topic.
+
+For a given query it:
+1. Searches Divepool for semantically relevant posts (with clustering)
+2. Enriches results with Bluesky engagement data (likes, reposts, replies)
+3. Resolves author profiles (follower counts, bios)
+4. Checks follow relationships between top authors to find community clusters
+5. Finds related posts by top authors to see what else they talk about
+6. Outputs a rich topic report
+"""
+
+import json
+import sys
+import time
+import urllib.request
+import urllib.error
+import urllib.parse
+from collections import defaultdict
+from concurrent.futures import ThreadPoolExecutor, as_completed
+
+DIVEPOOL = "https://divepool.social/api/v1/search"
+BSKY_API = "https://public.api.bsky.app/xrpc"
+
+
+# ── Divepool search ──────────────────────────────────────────────────────────
+
+def divepool_search(token: str, query: str, limit: int = 300) -> dict:
+    payload = json.dumps({
+        "query": query, "limit": limit,
+        "cluster": True, "distinct": True,
+    }).encode()
+    req = urllib.request.Request(DIVEPOOL, data=payload, headers={
+        "Content-Type": "application/json",
+        "Authorization": f"Bearer {token}",
+    }, method="POST")
+    with urllib.request.urlopen(req, timeout=30) as resp:
+        return json.loads(resp.read())
+
+
+# ── Bluesky public API helpers ───────────────────────────────────────────────
+
+def bsky_get(method: str, params: dict) -> dict | None:
+    qs = urllib.parse.urlencode(params)
+    url = f"{BSKY_API}/{method}?{qs}"
+    req = urllib.request.Request(url, headers={"Accept": "application/json"})
+    try:
+        with urllib.request.urlopen(req, timeout=10) as resp:
+            return json.loads(resp.read())
+    except urllib.error.HTTPError:
+        return None
+
+
+def get_profiles(dids: list[str]) -> dict[str, dict]:
+    """Batch-fetch profiles (up to 25 per call)."""
+    profiles = {}
+    for i in range(0, len(dids), 25):
+        batch = dids[i:i+25]
+        qs = "&".join(f"actors={urllib.parse.quote(d)}" for d in batch)
+        url = f"{BSKY_API}/app.bsky.actor.getProfiles?{qs}"
+        req = urllib.request.Request(url, headers={"Accept": "application/json"})
+        try:
+            with urllib.request.urlopen(req, timeout=10) as resp:
+                data = json.loads(resp.read())
+                for p in data.get("profiles", []):
+                    profiles[p["did"]] = p
+        except urllib.error.HTTPError:
+            pass
+    return profiles
+
+
+def get_post_thread(uri: str) -> dict | None:
+    return bsky_get("app.bsky.feed.getPostThread", {"uri": uri, "depth": 0})
+
+
+def get_follows(did: str, limit: int = 100) -> list[str]:
+    """Get DIDs that `did` follows."""
+    data = bsky_get("app.bsky.graph.getFollows", {"actor": did, "limit": limit})
+    if not data:
+        return []
+    return [f["did"] for f in data.get("follows", [])]
+
+
+# ── Main logic ───────────────────────────────────────────────────────────────
+
+def main():
+    if len(sys.argv) < 3:
+        print("Usage: python3 topic_explorer.py <bearer_token> <query>")
+        print('Example: python3 topic_explorer.py TOKEN "climate activism"')
+        sys.exit(1)
+
+    token = sys.argv[1]
+    query = " ".join(sys.argv[2:])
+
+    print(f"Topic Explorer: \"{query}\"")
+    print(f"{'─'*60}")
+
+    # ── Step 1: Semantic search ──────────────────────────────────────────
+    print("\n[1/5] Semantic search via Divepool...")
+    t0 = time.time()
+    search_data = divepool_search(token, query)
+    results = search_data.get("results", [])
+    clusters = search_data.get("clusters", [])
+    print(f"      {len(results)} results, {len(clusters)} clusters ({time.time()-t0:.1f}s)")
+
+    if not results:
+        print("No results found.")
+        return
+
+    # ── Step 2: Enrich with engagement data ──────────────────────────────
+    print("\n[2/5] Fetching engagement data from Bluesky...")
+    t0 = time.time()
+    top_results = results[:30]  # enrich top 30
+    engagement = {}
+
+    def fetch_engagement(r):
+        uri = f"at://{r['did']}/{r['collection']}/{r['rkey']}"
+        thread = get_post_thread(uri)
+        if thread and "thread" in thread:
+            post = thread["thread"].get("post", {})
+            return r["rkey"], {
+                "likes": post.get("likeCount", 0),
+                "reposts": post.get("repostCount", 0),
+                "replies": post.get("replyCount", 0),
+                "uri": uri,
+            }
+        return r["rkey"], None
+
+    with ThreadPoolExecutor(max_workers=10) as pool:
+        futures = [pool.submit(fetch_engagement, r) for r in top_results]
+        for f in as_completed(futures):
+            rkey, data = f.result()
+            if data:
+                engagement[rkey] = data
+
+    print(f"      Enriched {len(engagement)}/{len(top_results)} posts ({time.time()-t0:.1f}s)")
+
+    # ── Step 3: Resolve author profiles ──────────────────────────────────
+    print("\n[3/5] Resolving author profiles...")
+    t0 = time.time()
+    unique_dids = list(dict.fromkeys(r["did"] for r in top_results))[:25]
+    profiles = get_profiles(unique_dids)
+    print(f"      {len(profiles)} profiles resolved ({time.time()-t0:.1f}s)")
+
+    # ── Step 4: Check follow relationships between top authors ───────────
+    print("\n[4/5] Mapping follow graph between top authors...")
+    t0 = time.time()
+    top_dids = unique_dids[:15]  # check top 15
+    follow_graph: dict[str, set[str]] = {}
+    top_did_set = set(top_dids)
+
+    def fetch_follows(did):
+        follows = get_follows(did)
+        mutual = set(follows) & top_did_set
+        return did, mutual
+
+    with ThreadPoolExecutor(max_workers=8) as pool:
+        futures = [pool.submit(fetch_follows, did) for did in top_dids]
+        for f in as_completed(futures):
+            did, mutual = f.result()
+            if mutual - {did}:
+                follow_graph[did] = mutual - {did}
+
+    total_edges = sum(len(v) for v in follow_graph.values())
+    print(f"      {total_edges} follow links among top {len(top_dids)} authors ({time.time()-t0:.1f}s)")
+
+    # ── Step 5: Second-order search — what else do top authors post about?
+    print("\n[5/5] Probing what top authors talk about (second-order search)...")
+    t0 = time.time()
+    # Pick 3 contrasting queries based on cluster topics
+    alt_queries = []
+    for cl in sorted(clusters, key=lambda c: c["size"], reverse=True)[:3]:
+        topics = cl.get("topics", [])
+        if topics:
+            alt_queries.append(topics[0])
+
+    alt_results = {}
+    for aq in alt_queries:
+        try:
+            d = divepool_search(token, aq, limit=100)
+            alt_results[aq] = d.get("results", [])
+        except Exception:
+            pass
+    print(f"      Ran {len(alt_queries)} sub-queries ({time.time()-t0:.1f}s)")
+
+    # ── REPORT ───────────────────────────────────────────────────────────
+
+    def handle_for(did):
+        p = profiles.get(did, {})
+        return p.get("handle", did.split(":")[-1])
+
+    print(f"\n\n{'━'*60}")
+    print(f"  TOPIC REPORT: \"{query}\"")
+    print(f"{'━'*60}")
+
+    # Cluster overview
+    print(f"\n── Topic Clusters ──")
+    for cl in sorted(clusters, key=lambda c: c["size"], reverse=True):
+        topics = ", ".join(cl.get("topics", [])[:5])
+        print(f"\n  Cluster #{cl['id']}  ({cl['size']} posts)")
+        print(f"  Topics: {topics}")
+        # Sample posts from this cluster with engagement
+        shown = 0
+        for idx in cl.get("result_indices", [])[:3]:
+            if idx < len(results):
+                r = results[idx]
+                eng = engagement.get(r["rkey"])
+                eng_str = ""
+                if eng:
+                    eng_str = f"  [{eng['likes']}♥ {eng['reposts']}⟳ {eng['replies']}💬]"
+                text = r["text"][:100].replace("\n", " ")
+                print(f"    @{r.get('handle', '?'):25s}{eng_str}")
+                print(f"      \"{text}...\"")
+                shown += 1
+
+    # Top voices with profile context
+    print(f"\n── Top Voices ──")
+    scored = []
+    for r in top_results:
+        eng = engagement.get(r["rkey"], {})
+        impact = eng.get("likes", 0) + eng.get("reposts", 0) * 2
+        scored.append((r, impact))
+    scored.sort(key=lambda x: (-x[1], x[0]["score"]))
+
+    for r, impact in scored[:10]:
+        p = profiles.get(r["did"], {})
+        handle = p.get("handle", r.get("handle", "?"))
+        followers = p.get("followersCount", 0)
+        bio = (p.get("description") or "")[:80].replace("\n", " ")
+        eng = engagement.get(r["rkey"], {})
+        eng_str = ""
+        if eng:
+            eng_str = f"{eng['likes']}♥ {eng['reposts']}⟳ {eng['replies']}💬"
+
+        print(f"\n  @{handle}")
+        print(f"    {followers:,} followers | similarity={r['score']:.3f} | {eng_str}")
+        if bio:
+            print(f"    Bio: \"{bio}\"")
+        text = r["text"][:120].replace("\n", " ")
+        print(f"    Post: \"{text}...\"")
+
+    # Follow network
+    if follow_graph:
+        print(f"\n── Community Network (who follows whom among top authors) ──")
+        # Find clusters of mutual follows
+        for did, follows in sorted(follow_graph.items(), key=lambda x: -len(x[1])):
+            src = handle_for(did)
+            targets = ", ".join(f"@{handle_for(d)}" for d in follows)
+            print(f"  @{src} → {targets}")
+
+        # Identify the most-followed within the topic
+        in_degree: dict[str, int] = defaultdict(int)
+        for did, follows in follow_graph.items():
+            for f in follows:
+                in_degree[f] += 1
+        if in_degree:
+            print(f"\n  Hub accounts (most followed within topic):")
+            for did, count in sorted(in_degree.items(), key=lambda x: -x[1])[:5]:
+                print(f"    @{handle_for(did)} — followed by {count}/{len(top_dids)} top authors")
+
+    # Cross-topic presence
+    if alt_results:
+        print(f"\n── Related Conversations ──")
+        top_handles_main = {r.get("handle") for r in top_results}
+        for aq, ares in alt_results.items():
+            alt_handles = {r.get("handle") for r in ares}
+            overlap = top_handles_main & alt_handles - {None, ""}
+            print(f"\n  Sub-topic: \"{aq}\"")
+            if overlap:
+                print(f"    Overlapping voices: {', '.join('@'+h for h in list(overlap)[:5])}")
+            else:
+                print(f"    No overlapping voices (distinct sub-community)")
+            # Show top post from this sub-topic
+            if ares:
+                r = ares[0]
+                text = r["text"][:120].replace("\n", " ")
+                print(f"    Top hit: @{r.get('handle', '?')}: \"{text}...\"")
+
+    print(f"\n{'━'*60}")
+    print(f"  Done. {len(results)} posts → {len(clusters)} clusters → "
+          f"{len(profiles)} profiles → {total_edges} follow links")
+    print(f"{'━'*60}\n")
+
+
+if __name__ == "__main__":
+    main()
diff --git a/experiments/vibe_map.py b/experiments/vibe_map.py
new file mode 100644
index 0000000..05c00f9
--- /dev/null
+++ b/experiments/vibe_map.py
@@ -0,0 +1,138 @@
+#!/usr/bin/env python3
+"""
+Bluesky Vibe Map — semantic pulse of the network via Divepool search.
+
+Runs a batch of diverse queries against the Divepool search API with
+clustering enabled, then prints a compact digest: top clusters per query,
+trending handles, and cross-query topic overlaps.
+"""
+
+import json
+import sys
+import time
+import urllib.request
+
+BASE_URL = "https://divepool.social/api/v1/search"
+
+QUERIES = [
+    "climate change action protest",
+    "open source software community",
+    "mental health support",
+    "indie game development",
+    "astronomy astrophotography space",
+    "cooking recipes food",
+    "live music concerts festival",
+    "labor union strike workers",
+    "generative AI art ethics",
+    "queer joy pride community",
+]
+
+def search(token: str, query: str, limit: int = 200) -> dict:
+    payload = json.dumps({
+        "query": query,
+        "limit": limit,
+        "cluster": True,
+        "distinct": True,
+    }).encode()
+    req = urllib.request.Request(
+        BASE_URL,
+        data=payload,
+        headers={
+            "Content-Type": "application/json",
+            "Authorization": f"Bearer {token}",
+        },
+        method="POST",
+    )
+    with urllib.request.urlopen(req, timeout=30) as resp:
+        return json.loads(resp.read())
+
+
+def print_query_digest(query: str, data: dict, elapsed: float):
+    results = data.get("results", [])
+    clusters = data.get("clusters", [])
+
+    print(f"\n{'='*60}")
+    print(f"  \"{query}\"")
+    print(f"  {len(results)} results, {len(clusters)} clusters, {elapsed:.1f}s")
+    print(f"{'='*60}")
+
+    # Show top clusters sorted by size
+    for cl in sorted(clusters, key=lambda c: c["size"], reverse=True)[:5]:
+        topics = ", ".join(cl.get("topics", [])[:4])
+        # Grab a sample post from this cluster
+        sample_idx = cl["result_indices"][0] if cl["result_indices"] else None
+        sample_text = ""
+        if sample_idx is not None and sample_idx < len(results):
+            sample_text = results[sample_idx]["text"][:120].replace("\n", " ")
+        print(f"\n  Cluster #{cl['id']}  ({cl['size']} posts)")
+        print(f"    Topics: {topics}")
+        if sample_text:
+            print(f"    Sample: \"{sample_text}...\"")
+
+    # Top handles by score
+    top = sorted(results[:20], key=lambda r: r["score"])[:5]
+    print(f"\n  Top handles:")
+    for r in top:
+        print(f"    @{r['handle']:30s}  score={r['score']:.3f}")
+
+
+def main():
+    if len(sys.argv) < 2:
+        print("Usage: python3 vibe_map.py <bearer_token> [query ...]")
+        sys.exit(1)
+
+    token = sys.argv[1]
+    queries = sys.argv[2:] if len(sys.argv) > 2 else QUERIES
+
+    all_topics: dict[str, list[str]] = {}  # topic -> list of queries it appeared in
+    all_handles: dict[str, int] = {}       # handle -> count across queries
+
+    print("Bluesky Vibe Map")
+    print(f"Scanning {len(queries)} queries...\n")
+
+    for query in queries:
+        t0 = time.time()
+        try:
+            data = search(token, query)
+        except Exception as e:
+            print(f"\n  ERROR on \"{query}\": {e}")
+            continue
+        elapsed = time.time() - t0
+
+        print_query_digest(query, data, elapsed)
+
+        # Accumulate cross-query stats
+        for cl in data.get("clusters", []):
+            for topic in cl.get("topics", []):
+                all_topics.setdefault(topic, []).append(query)
+        for r in data.get("results", []):
+            h = r.get("handle", "")
+            if h:
+                all_handles[h] = all_handles.get(h, 0) + 1
+
+    # Cross-query summary
+    print(f"\n\n{'#'*60}")
+    print(f"  CROSS-QUERY SUMMARY")
+    print(f"{'#'*60}")
+
+    # Topics appearing across multiple queries
+    shared = {t: qs for t, qs in all_topics.items() if len(qs) > 1}
+    if shared:
+        print(f"\n  Topics spanning multiple queries:")
+        for topic, qs in sorted(shared.items(), key=lambda x: -len(x[1]))[:10]:
+            print(f"    \"{topic}\" — appears in {len(qs)} queries")
+    else:
+        print(f"\n  No topics shared across queries (clusters are well-separated)")
+
+    # Handles appearing in multiple queries (cross-topic posters)
+    multi = {h: c for h, c in all_handles.items() if c > 1}
+    if multi:
+        print(f"\n  Cross-topic voices (appear in 2+ queries):")
+        for h, c in sorted(multi.items(), key=lambda x: -x[1])[:15]:
+            print(f"    @{h} — {c} queries")
+
+    print()
+
+
+if __name__ == "__main__":
+    main()