[eric] onboarding: harvest real chat history at first run via browser-session cookies (Claude proven live)

backend/apps/onboarding/onboarding.py

@@ -43,13 +43,20 @@ def post_scan() -> dict:
 @onboarding.router.post("/prep")
 @typechecked
 async def post_prep(body: PrepRequest) -> dict:
-    from backend.apps.onboarding.chatgpt_usage import harvest_chatgpt_usage
+    from backend.apps.onboarding.usage.chatgpt_usage import harvest_chatgpt_usage
+    from backend.apps.onboarding.usage.claude_usage import harvest_claude_usage
     from backend.apps.settings.store import load_settings
 
-    # The frontend read needs a logged-in provider CARD, which a fresh install lacks; the codex connect token reads the ChatGPT backend directly, so fill the gap here.
+    # The frontend read needs a logged-in provider CARD, which a fresh install lacks. Fill the gap from what we already have: ChatGPT via the codex connect token, Claude via the user's own browser session cookies. Each fails open to "", so a missing one just drops out.
     if not body.usage_summary.strip():
-        harvested = await harvest_chatgpt_usage()
-        if harvested:
-            body.usage_summary = harvested
+        parts: list[str] = []
+        chatgpt = await harvest_chatgpt_usage()
+        if chatgpt:
+            parts.append("ChatGPT usage:\n" + chatgpt)
+        claude = await harvest_claude_usage()
+        if claude:
+            parts.append("Claude usage:\n" + claude)
+        if parts:
+            body.usage_summary = "\n\n".join(parts)
 
     return (await build_prep(load_settings(), body)).model_dump()

backend/apps/onboarding/usage/browser_cookies.py

@@ -0,0 +1,173 @@
+"""Read the user's own logged-in provider cookies from their real browser, so
+onboarding can harvest their actual chat history at first run without an in-app login.
+
+macOS + Chromium only for now (Chrome/Arc/Brave/Edge). We first find WHICH store holds
+the session by counting cookie names in the SQLite (no decryption, no keychain), then
+decrypt only that one store, so the "Safe Storage" keychain is touched at most once per
+browser (cached for the process). Values are v10/v11 AES-CBC. Fails open to {} on
+anything (no browser, app-bound v20 cookies, denied keychain, Safari-only user), so
+prep just falls back to the local scan.
+
+Only ever reads the specific provider domain asked for; never a general cookie sweep.
+The values are session secrets: used in-process for the harvest, never logged or stored.
+"""
+
+import hashlib
+import os
+import shutil
+import sqlite3
+import subprocess
+import tempfile
+from typing import Dict, List, Optional, Tuple
+
+from typeguard import typechecked
+
+CHROMIUM_ROOTS = {
+    "Chrome": "Library/Application Support/Google/Chrome",
+    "Arc": "Library/Application Support/Arc/User Data",
+    "Brave": "Library/Application Support/BraveSoftware/Brave-Browser",
+    "Edge": "Library/Application Support/Microsoft Edge",
+}
+KEYCHAIN_SERVICE = {
+    "Chrome": "Chrome Safe Storage",
+    "Arc": "Arc Safe Storage",
+    "Brave": "Brave Safe Storage",
+    "Edge": "Microsoft Edge Safe Storage",
+}
+PROFILES = ["Default"] + [f"Profile {i}" for i in range(1, 12)]
+
+# One keychain read per browser per process; "Always Allow" then never re-prompts.
+p_key_cache: Dict[str, Optional[bytes]] = {}
+
+
+@typechecked
+def p_safe_storage_key(browser: str) -> Optional[bytes]:
+    if browser in p_key_cache:
+        return p_key_cache[browser]
+    key: Optional[bytes] = None
+    try:
+        r = subprocess.run(
+            ["security", "find-generic-password", "-w", "-s", KEYCHAIN_SERVICE[browser]],
+            capture_output=True, text=True, timeout=20,
+        )
+        pw = r.stdout.strip()
+        if pw:
+            key = hashlib.pbkdf2_hmac("sha1", pw.encode(), b"saltysalt", 1003, 16)
+    except Exception:
+        key = None
+    p_key_cache[browser] = key
+    return key
+
+
+@typechecked
+def p_count_domain(db_path: str, domain: str) -> int:
+    tmp = tempfile.mktemp()
+    try:
+        shutil.copy2(db_path, tmp)
+        con = sqlite3.connect(f"file:{tmp}?mode=ro", uri=True)
+        cur = con.cursor()
+        cur.execute("SELECT count(*) FROM cookies WHERE host_key LIKE ?", (f"%{domain}",))
+        n = int(cur.fetchone()[0])
+        con.close()
+        return n
+    except Exception:
+        return 0
+    finally:
+        try:
+            os.remove(tmp)
+        except OSError:
+            pass
+
+
+@typechecked
+def p_best_store(domain: str) -> Optional[Tuple[str, str]]:
+    """The (browser, db_path) holding the most cookies for `domain`, found WITHOUT the keychain."""
+    home = os.path.expanduser("~")
+    best: Optional[Tuple[str, str]] = None
+    best_score = (0, -1.0)
+    for browser, rel in CHROMIUM_ROOTS.items():
+        base = os.path.join(home, rel)
+        if not os.path.isdir(base):
+            continue
+        for prof in PROFILES:
+            for sub in ("Cookies", "Network/Cookies"):
+                path = os.path.join(base, prof, sub)
+                if not os.path.isfile(path):
+                    continue
+                n = p_count_domain(path, domain)
+                if n:
+                    score = (n, os.path.getmtime(path))
+                    if score > best_score:
+                        best, best_score = (browser, path), score
+    return best
+
+
+@typechecked
+def p_decrypt(enc: bytes, key: bytes) -> Optional[str]:
+    if enc[:3] not in (b"v10", b"v11"):
+        return None  # v20 = app-bound encryption, out of reach without the browser
+    try:
+        from cryptography.hazmat.backends import default_backend
+        from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes
+
+        c = Cipher(algorithms.AES(key), modes.CBC(b" " * 16), backend=default_backend())
+        d = c.decryptor()
+        dec = d.update(enc[3:]) + d.finalize()
+        dec = dec[: -dec[-1]]  # strip PKCS7 padding
+        for cut in (0, 32):  # newer Chromium prepends a 32-byte domain hash
+            try:
+                return dec[cut:].decode("utf-8")
+            except UnicodeDecodeError:
+                continue
+    except Exception:
+        return None
+    return None
+
+
+@typechecked
+def read_provider_cookies(domain: str) -> Dict[str, str]:
+    """Decrypted cookie jar for `domain`, from whichever browser store actually has the session. At most one keychain touch (that store's browser), cached for the process."""
+    store = p_best_store(domain)
+    if store is None:
+        return {}
+    browser, db_path = store
+    key = p_safe_storage_key(browser)
+    if key is None:
+        return {}
+    jar: Dict[str, str] = {}
+    tmp = tempfile.mktemp()
+    try:
+        shutil.copy2(db_path, tmp)
+        con = sqlite3.connect(f"file:{tmp}?mode=ro", uri=True)
+        cur = con.cursor()
+        cur.execute("SELECT name, encrypted_value FROM cookies WHERE host_key LIKE ?", (f"%{domain}",))
+        for name, enc in cur.fetchall():
+            if not enc:
+                continue
+            val = p_decrypt(bytes(enc), key)
+            if val:
+                jar[str(name)] = val
+        con.close()
+    except Exception:
+        pass
+    finally:
+        try:
+            os.remove(tmp)
+        except OSError:
+            pass
+    return jar
+
+
+@typechecked
+def cookie_header(jar: Dict[str, str]) -> str:
+    return "; ".join(f"{k}={v}" for k, v in jar.items())
+
+
+@typechecked
+def logged_in_providers() -> List[str]:
+    """Which providers have a readable session, WITHOUT decrypting or touching the keychain: safe for a UI presence check."""
+    out: List[str] = []
+    for provider, domain in (("codex", "chatgpt.com"), ("claude", "claude.ai"), ("gemini", "gemini.google.com")):
+        if p_best_store(domain) is not None:
+            out.append(provider)
+    return out

backend/apps/onboarding/usage/claude_usage.py

@@ -0,0 +1,76 @@
+"""Read the user's real Claude conversation topics from claude.ai using their own
+logged-in browser cookies (see browser_cookies), no in-app login. Claude's website
+session is the only way in (its API token is a different realm), and a plain request
+carries it fine (unlike ChatGPT, claude.ai does not fingerprint-block). Capped,
+read-only, fails open to "" on anything so prep falls back to the local scan.
+"""
+
+from typing import List
+
+import httpx
+from typeguard import typechecked
+
+from backend.apps.onboarding.usage.browser_cookies import cookie_header, read_provider_cookies
+
+BASE = "https://claude.ai"
+UA = "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/126.0.0.0 Safari/537.36"
+PAGE = 100
+CAP_PAGES = 40
+CAP_TITLES = 1000
+
+
+@typechecked
+def summarize_claude_usage(total: int, titles: List[str]) -> str:
+    parts: List[str] = []
+    if total > 0:
+        parts.append(f"They have {total} past Claude conversations.")
+    if titles:
+        parts.append("Topics they keep coming back to (recent first): " + "; ".join(titles[:150]))
+    return "\n".join(parts)[:4000]
+
+
+@typechecked
+async def harvest_claude_usage() -> str:
+    jar = read_provider_cookies("claude.ai")
+    if not jar:
+        return ""
+    headers = {"Cookie": cookie_header(jar), "User-Agent": UA, "Accept": "application/json"}
+    titles: List[str] = []
+    seen: set = set()
+    try:
+        async with httpx.AsyncClient(timeout=15.0, headers=headers) as client:
+            org_res = await client.get(f"{BASE}/api/organizations")
+            if org_res.status_code != 200:
+                return ""
+            orgs = org_res.json()
+            if not isinstance(orgs, list) or not orgs:
+                return ""
+            org = orgs[0].get("uuid")
+            offset = 0
+            for _ in range(CAP_PAGES):
+                if len(titles) >= CAP_TITLES:
+                    break
+                cr = await client.get(
+                    f"{BASE}/api/organizations/{org}/chat_conversations",
+                    params={"limit": PAGE, "offset": offset},
+                )
+                if cr.status_code != 200:
+                    break
+                items = cr.json()
+                if not isinstance(items, list) or not items:
+                    break
+                fresh = 0
+                for it in items:
+                    cid = it.get("uuid")
+                    if cid and cid not in seen:
+                        seen.add(cid)
+                        name = it.get("name")
+                        if name:
+                            titles.append(str(name))
+                        fresh += 1
+                if fresh == 0 or len(items) < PAGE:
+                    break
+                offset += PAGE
+    except Exception:
+        return ""
+    return summarize_claude_usage(len(seen), titles)

backend/tests/test_onboarding.py

@@ -114,7 +114,7 @@ def test_parse_prep_carries_reasons():
 
 
 def test_summarize_chatgpt_usage_leads_with_memory_and_caps():
-    from backend.apps.onboarding.chatgpt_usage import summarize_chatgpt_usage
+    from backend.apps.onboarding.usage.chatgpt_usage import summarize_chatgpt_usage
 
     s = summarize_chatgpt_usage(812, ["Has an Akita", "Squats 495"], ["Swift concurrency", "Deadlift form"])
     assert "812 past AI conversations" in s
@@ -126,12 +126,38 @@ def test_summarize_chatgpt_usage_leads_with_memory_and_caps():
 
 @pytest.mark.asyncio
 async def test_harvest_chatgpt_usage_fails_open_without_codex(monkeypatch):
-    from backend.apps.onboarding import chatgpt_usage
+    from backend.apps.onboarding.usage import chatgpt_usage
 
     monkeypatch.setattr(chatgpt_usage, "read_persisted_connections", lambda: [])
     assert await chatgpt_usage.harvest_chatgpt_usage() == ""
 
 
+def test_read_provider_cookies_fails_open_without_a_store(monkeypatch):
+    from backend.apps.onboarding.usage import browser_cookies
+
+    # No browser store has the domain -> empty jar, and the keychain is never touched.
+    monkeypatch.setattr(browser_cookies, "p_best_store", lambda domain: None)
+    assert browser_cookies.read_provider_cookies("claude.ai") == {}
+
+
+def test_summarize_claude_usage_counts_and_caps():
+    from backend.apps.onboarding.usage.claude_usage import summarize_claude_usage
+
+    s = summarize_claude_usage(490, ["Yuji Itadori and Buddhism", "B2B SaaS Startup Ideas"])
+    assert "490 past Claude conversations" in s
+    assert "Yuji Itadori and Buddhism; B2B SaaS Startup Ideas" in s
+    big = summarize_claude_usage(1000, [f"topic number {i} about something specific" for i in range(1000)])
+    assert len(big) <= 4000
+
+
+@pytest.mark.asyncio
+async def test_harvest_claude_usage_fails_open_without_cookies(monkeypatch):
+    from backend.apps.onboarding.usage import claude_usage
+
+    monkeypatch.setattr(claude_usage, "read_provider_cookies", lambda domain: {})
+    assert await claude_usage.harvest_claude_usage() == ""
+
+
 @pytest.mark.asyncio
 async def test_build_prep_fails_open_without_provider(monkeypatch):
     async def boom(*args, **kwargs):