Phase 0 + A: capability probe, SQLite storage + HF login

.gitignore

@@ -59,6 +59,7 @@ gradio_cached_examples/
 # Generated adventures / sites, model caches, local DBs
 generated_sites/
 adventures/
+.frogquest/        # local dev SQLite (store.py fallback when /data isn't mounted) + WAL sidecars
 *.gguf
 *.sqlite
 *.db

README.md

@@ -7,6 +7,7 @@ sdk: gradio
 sdk_version: 6.17.3
 python_version: '3.12'
 app_file: app.py
+hf_oauth: true
 pinned: true
 license: apache-2.0
 short_description: A text-image based RPG built to help fight procrastination
@@ -26,8 +27,11 @@ boss quest you face first.
 - **FLUX.2 [klein] 4B** (diffusers) generates each quest's image using your uploaded photo as a
   reference, then edits that same image into success / "retreat to fight another day" states.
 
-Your photo and quest state live **only in your browser** (localStorage); the photo is sent to the
-GPU transiently for generation and is never stored server-side.
+Logged out, your photo and quest state live **only in your browser** (localStorage). When you
+**log in with your Hugging Face account**, the same state — photo, quests, campaigns, and generated
+images — is also saved to your **private per-account record** (SQLite on the Space's persistent
+storage) so it survives a cleared browser. The photo is sent to the GPU only transiently during
+generation.
 
 ### GPU backend (ZeroGPU or Modal)
 

app.py

@@ -9,8 +9,9 @@ Layout (one page, no onboarding gate):
 
 State lives in gr.BrowserState (per-browser localStorage): the resized photo (base64), the
 chosen world/theme, the validated adventure/quests JSON, the selected quest id, and a cache of
-generated scene images (JPEG data-urls). The photo goes to the GPU only transiently during
-generation and is never persisted server-side.
+generated scene images (JPEG data-urls). When you log in with an HF account, the SAME full state
+(photo and generated images included) is also saved to your private per-account record in SQLite
+(store.py) so it survives a cleared browser; logged-out use stays browser-only.
 
 Images generate LAZILY: selecting a quest with no cached scene generates it on the spot; Done /
 Couldn't EDIT the cached initial scene into a success / failure state (never regenerate).
@@ -32,6 +33,7 @@ import spaces
 from PIL import Image
 
 from schema import THEMES, merge_quests, validate_and_clamp
+import store
 
 
 # ZeroGPU scans for a @spaces.GPU function at startup; this guarantees detection regardless
@@ -403,10 +405,24 @@ def reset_all():
             _desc_html(None, None), _stats_html([]), "fantasy")
 
 
-def boot(browser):
-    """Restore everything from BrowserState on page load. Shows cached scenes only (never
-    generates) so reloads are instant; uncached scenes regenerate when the quest is clicked."""
-    b = browser or _default_state()
+def persist(uid, browser):
+    """Write-through the full state to the user's durable SQLite record. No-op when logged out
+    (anonymous users live in BrowserState only). Bound via .then(...) after every write event."""
+    store.save(uid, browser)
+
+
+def boot(browser, request: gr.Request):
+    """Restore on page load. If logged in (HF OAuth) the durable SQLite record is the source of
+    truth and is mirrored back into BrowserState; otherwise BrowserState is used, and migrated into
+    SQLite on first login. Shows cached scenes only (never generates) so reloads are instant."""
+    uid = getattr(request, "username", None) if request is not None else None
+    saved = store.load(uid)
+    if saved:
+        b = saved
+    else:
+        b = browser or _default_state()
+        if uid:
+            store.save(uid, b)  # first login on this account: adopt the browser's current state
     photo = b.get("photo")
     quests = b.get("quests") or []
     adventure = b.get("adventure")
@@ -422,7 +438,7 @@ def boot(browser):
         if data:
             scene = _dataurl_to_pil(data)
     photo_pil = _dataurl_to_pil(photo) if photo else None
-    return (photo_pil, photo, quests, adventure, selected, images,
+    return (uid, b, photo_pil, photo, quests, adventure, selected, images,
             scene, _desc_html(quest, adventure), _stats_html(quests), theme)
 
 
@@ -439,8 +455,11 @@ with gr.Blocks(title="FrogQuest") as demo:
     adventure_state = gr.State(None)      # title / art_style / seed / theme
     selected_id_state = gr.State(None)    # currently selected quest id
     images_state = gr.State({})           # {quest_id: {initial/success/failure: jpeg data-url}}
+    uid_state = gr.State(None)            # HF login id (None = anonymous); set by boot from gr.Request
 
-    gr.HTML('<div class="fq-topbar"><h1 class="fq-logo">FROG<b>QUEST</b></h1></div>')
+    with gr.Row(elem_classes=["fq-topbar"]):
+        gr.HTML('<h1 class="fq-logo">FROG<b>QUEST</b></h1>')
+        login_btn = gr.LoginButton(elem_classes=["fq-login"])
 
     with gr.Row(elem_classes=["fq-app-grid"]):
         # ---------- LEFT: hero ----------
@@ -450,8 +469,9 @@ with gr.Blocks(title="FrogQuest") as demo:
                 type="pil", sources=["upload"], show_label=False, height=190,
                 elem_classes=["fq-photo"],
             )
-            gr.HTML('<p class="hint">Click to upload / change your photo. It stays in your '
-                    'browser — sent to the GPU only to draw your scenes, never stored.</p>')
+            gr.HTML('<p class="hint">Click to upload / change your photo. Sent to the GPU only to '
+                    'draw your scenes. Stored in your browser, and in your private account record '
+                    'when you log in.</p>')
             stats_html = gr.HTML(_stats_html([]))
             gr.HTML('<h3 class="panel-subtitle">WORLD</h3>')
             theme_radio = gr.Radio(
@@ -512,7 +532,7 @@ with gr.Blocks(title="FrogQuest") as demo:
                                               select_quest(qid, photo, adv, imgs, qs, br)))(q["id"]),
                                 inputs=[photo_state, adventure_state, images_state, quests_state, browser],
                                 outputs=[selected_id_state, scene_image, desc_html, images_state, browser],
-                            )
+                            ).then(persist, [uid_state, browser])
                             # Clear button appears only once a quest is done/failed, to remove it.
                             if finished:
                                 clr = gr.Button("✕", elem_classes=["pix-btn", "small", "fq-task-clear"])
@@ -522,7 +542,7 @@ with gr.Blocks(title="FrogQuest") as demo:
                                     inputs=[quests_state, adventure_state, images_state, selected_id_state, browser],
                                     outputs=[quests_state, images_state, selected_id_state, scene_image,
                                              desc_html, stats_html, browser],
-                                )
+                                ).then(persist, [uid_state, browser])
 
     # ---------- BOTTOM: Frog Master chat ----------
     with gr.Row(elem_classes=["fq-chatbar"]):
@@ -535,36 +555,37 @@ with gr.Blocks(title="FrogQuest") as demo:
             '"I finished the report" · "Couldn\'t do the gym — too tired"</p><p>Where to Start, Where to Break and let Go,<br>Where to change <br>Where to look and to grow </p>')
 
     # ---------- wiring ----------
-    photo_image.upload(upload_photo, [photo_image, browser], [photo_state, browser])
+    photo_image.upload(upload_photo, [photo_image, browser], [photo_state, browser]
+                       ).then(persist, [uid_state, browser])
     theme_radio.change(
         change_theme, [theme_radio, adventure_state, images_state, browser],
         [adventure_state, images_state, scene_image, desc_html, browser],
-    )
+    ).then(persist, [uid_state, browser])
     reset_btn.click(
         reset_all, None,
         [browser, photo_image, photo_state, quests_state, adventure_state, selected_id_state,
          images_state, scene_image, desc_html, stats_html, theme_radio],
-    )
+    ).then(persist, [uid_state, browser])
     done_btn.click(
         apply_done, [selected_id_state, quests_state, adventure_state, photo_state, images_state, browser],
         [quests_state, images_state, scene_image, desc_html, stats_html, browser],
-    )
+    ).then(persist, [uid_state, browser])
     couldnt_btn.click(
         apply_couldnt, [selected_id_state, quests_state, adventure_state, photo_state, images_state, browser],
         [quests_state, images_state, scene_image, desc_html, stats_html, browser],
-    )
+    ).then(persist, [uid_state, browser])
 
     _chat_inputs = [chat_input, quests_state, adventure_state, theme_radio, photo_state,
                     selected_id_state, images_state, browser]
     _chat_outputs = [chat_input, quests_state, adventure_state, selected_id_state, images_state,
                      scene_image, desc_html, stats_html, browser]
-    send_btn.click(chat_send, _chat_inputs, _chat_outputs)
-    chat_input.submit(chat_send, _chat_inputs, _chat_outputs)
+    send_btn.click(chat_send, _chat_inputs, _chat_outputs).then(persist, [uid_state, browser])
+    chat_input.submit(chat_send, _chat_inputs, _chat_outputs).then(persist, [uid_state, browser])
 
     demo.load(
         boot, [browser],
-        [photo_image, photo_state, quests_state, adventure_state, selected_id_state, images_state,
-         scene_image, desc_html, stats_html, theme_radio],
+        [uid_state, browser, photo_image, photo_state, quests_state, adventure_state,
+         selected_id_state, images_state, scene_image, desc_html, stats_html, theme_radio],
     )
 
 

probe_nemotron.py

@@ -0,0 +1,252 @@
+"""Phase 0 capability probe (THROWAWAY — not part of the app, safe to delete).
+
+Question it answers: how capable is Nemotron Nano 4B at the *narrow* language jobs the
+"thin LLM, fat code" plan leaves to it? We do NOT test scheduling/structure here — that's
+deterministic Python. We test only:
+
+  1. intent  — classify a user message into the router enum (reuses the app's real prompt/schema)
+  2. ask     — turn a long-term goal into 1-3 clarifying questions
+  3. plan    — turn a goal + web-research snippets into an ordered milestone list
+  4. quest   — turn one milestone into themed quest title + narrative
+
+For each task we report: did it return VALID JSON to the schema, and a light quality heuristic.
+Eyeball the printed outputs — the pass-rate is a guardrail, the text quality is the real signal.
+
+Run:
+    python probe_nemotron.py            # auto: GPU if available, else CPU (slow, ~4.3GB download)
+    FROGQUEST_PROBE_GPU_LAYERS=-1 python probe_nemotron.py   # force all layers on GPU
+    FROGQUEST_PROBE_GPU_LAYERS=0  python probe_nemotron.py   # force CPU
+
+Reuses gpu_shared (model id/config, extract_json, CUDA preloader) and schema (INTENT_SCHEMA)
+so the probe exercises the same path the app will.
+"""
+from __future__ import annotations
+
+import importlib.util
+import os
+import time
+
+# Faster downloads only if hf_transfer is actually installed — setting the flag without the
+# package makes huggingface_hub raise on download (common on a fresh local CPU box).
+if os.environ.get("HF_HUB_ENABLE_HF_TRANSFER") is None and importlib.util.find_spec("hf_transfer"):
+    os.environ["HF_HUB_ENABLE_HF_TRANSFER"] = "1"
+
+from gpu_shared import (  # noqa: E402
+    GGUF_FILE,
+    GGUF_REPO,
+    INTENT_SYSTEM_PROMPT,
+    extract_json,
+    preload_cuda_libs,
+)
+from schema import INTENT_SCHEMA  # noqa: E402
+
+
+# ----------------------------- model -----------------------------
+
+def _gpu_layers() -> int:
+    """All layers on GPU when one is available, else CPU. Override with FROGQUEST_PROBE_GPU_LAYERS."""
+    override = os.environ.get("FROGQUEST_PROBE_GPU_LAYERS")
+    if override is not None:
+        return int(override)
+    try:
+        import torch
+        return -1 if torch.cuda.is_available() else 0
+    except Exception:
+        return 0
+
+
+def load_llm():
+    layers = _gpu_layers()
+    print(f"[probe] loading {GGUF_REPO} ({GGUF_FILE}) with n_gpu_layers={layers} ...")
+    try:
+        preload_cuda_libs()  # no-op without the CUDA libs (e.g. pure-CPU local run)
+    except Exception:
+        pass
+    from llama_cpp import Llama
+    t0 = time.time()
+    llm = Llama.from_pretrained(
+        repo_id=GGUF_REPO, filename=GGUF_FILE,
+        n_gpu_layers=layers, n_ctx=4096, verbose=False,
+    )
+    print(f"[probe] model ready in {time.time() - t0:.1f}s\n")
+    return llm
+
+
+def ask(llm, system: str, user: str, schema: dict, max_tokens: int = 512):
+    """One constrained call. Returns (raw_text, parsed_or_None, seconds)."""
+    t0 = time.time()
+    try:
+        out = llm.create_chat_completion(
+            messages=[{"role": "system", "content": system},
+                      {"role": "user", "content": user}],
+            response_format={"type": "json_object", "schema": schema},
+            temperature=0.0, max_tokens=max_tokens,
+        )
+        raw = out["choices"][0]["message"]["content"]
+    except Exception as e:  # generation itself blew up
+        return f"<<ERROR: {e}>>", None, time.time() - t0
+    parsed = extract_json(raw)
+    return raw, (parsed if isinstance(parsed, dict) and parsed else None), time.time() - t0
+
+
+# ----------------------------- throwaway prompts/schemas for the NEW tasks -----------------------------
+
+ASK_SYS = (
+    "You help plan a user's long-term goal. Given the goal, output the 1-3 most important "
+    "clarifying questions you must ask before you could plan it (deadline, current level, time "
+    "available, constraints). OUTPUT JSON ONLY: {\"questions\": [\"...\"]}. No prose.\n/no_think"
+)
+ASK_SCHEMA = {
+    "type": "object",
+    "properties": {"questions": {"type": "array", "items": {"type": "string"}}},
+    "required": ["questions"],
+}
+
+PLAN_SYS = (
+    "You turn a long-term goal plus web-research snippets into an ordered plan. OUTPUT JSON ONLY: "
+    "{\"milestones\": [{\"title\": \"\", \"summary\": \"\"}]} with 3-6 milestones in the order "
+    "they should be done. Ground them in the snippets. No prose.\n/no_think"
+)
+PLAN_SCHEMA = {
+    "type": "object",
+    "properties": {
+        "milestones": {
+            "type": "array",
+            "items": {
+                "type": "object",
+                "properties": {"title": {"type": "string"}, "summary": {"type": "string"}},
+                "required": ["title", "summary"],
+            },
+        }
+    },
+    "required": ["milestones"],
+}
+
+QUEST_SYS = (
+    "You are a quest writer. Given one real task/milestone and a theme, write a short themed quest "
+    "title and a 1-2 sentence narrative where the user is the hero. OUTPUT JSON ONLY: "
+    "{\"quest_title\": \"\", \"narrative\": \"\"}. No prose.\n/no_think"
+)
+QUEST_SCHEMA = {
+    "type": "object",
+    "properties": {"quest_title": {"type": "string"}, "narrative": {"type": "string"}},
+    "required": ["quest_title", "narrative"],
+}
+
+
+# ----------------------------- test batteries -----------------------------
+# Each case: (user_input, checker(parsed)->bool, label). The checker is a LIGHT quality heuristic
+# on top of "valid JSON"; the printed text is the real signal.
+
+INTENT_CASES = [
+    ("Finish the tax report, reply to emails, book the dentist", "forge"),
+    ("I finished the tax report", "mark_done"),
+    ("Couldn't make it to the gym, too tired", "mark_couldnt"),
+    ("Also add: call mom and water the plants", "add_tasks"),
+    ("what's the weather like today?", "unknown"),
+    ("start over, here are my plans for the week", "forge"),
+]
+
+ASK_CASES = [
+    "I want to score in the 95th percentile on the GRE",
+    "Help me win this hackathon",
+    "I want to learn to play guitar",
+    "Get fit",
+    "Launch my SaaS side project",
+]
+
+PLAN_CASES = [
+    ("Run a 10k race", [
+        "Couch-to-10k programs build from run/walk intervals over 8-10 weeks.",
+        "Most plans run 3-4 days/week with one long run that grows weekly.",
+        "Rest days and easy pace prevent injury; add a tempo run after week 4.",
+    ]),
+    ("Learn basic Spanish for travel", [
+        "Start with the 500 most common words and survival phrases.",
+        "Apps like Anki use spaced repetition for vocab retention.",
+        "Practicing speaking early, even badly, accelerates fluency.",
+    ]),
+]
+
+QUEST_CASES = [
+    ("Write the introduction section of the thesis", "fantasy"),
+    ("Do 20 minutes of cardio", "cyberpunk"),
+    ("Refactor the auth module", "space"),
+]
+
+
+def run_intent(llm):
+    print("=== TASK 1: intent classification (real app prompt + schema) ===")
+    ok_json = ok_label = 0
+    for msg, expected in INTENT_CASES:
+        raw, parsed, secs = ask(llm, INTENT_SYSTEM_PROMPT, f"Context:\nNo quest log exists yet.\n\nUser message:\n{msg}",
+                                INTENT_SCHEMA, max_tokens=128)
+        got = (parsed or {}).get("intent")
+        ok_json += parsed is not None
+        match = got == expected
+        ok_label += match
+        print(f"  [{secs:4.1f}s] json={'Y' if parsed else 'N'} got={got!r:16} want={expected!r:14} {'OK' if match else 'MISS'} | {msg[:48]}")
+    print(f"  -> valid-JSON {ok_json}/{len(INTENT_CASES)} | correct-intent {ok_label}/{len(INTENT_CASES)}\n")
+
+
+def run_ask(llm):
+    print("=== TASK 2: goal -> 1-3 clarifying questions ===")
+    ok_json = ok_quality = 0
+    for goal in ASK_CASES:
+        raw, parsed, secs = ask(llm, ASK_SYS, f"Goal: {goal}", ASK_SCHEMA, max_tokens=256)
+        qs = (parsed or {}).get("questions") if parsed else None
+        good = isinstance(qs, list) and 1 <= len(qs) <= 3 and all(isinstance(q, str) and q.strip() for q in qs)
+        ok_json += parsed is not None
+        ok_quality += good
+        preview = " / ".join(qs)[:90] if isinstance(qs, list) else "(none)"
+        print(f"  [{secs:4.1f}s] json={'Y' if parsed else 'N'} 1-3q={'Y' if good else 'N'} | {goal[:30]:30} -> {preview}")
+    print(f"  -> valid-JSON {ok_json}/{len(ASK_CASES)} | 1-3 good questions {ok_quality}/{len(ASK_CASES)}\n")
+
+
+def run_plan(llm):
+    print("=== TASK 3: goal + snippets -> ordered milestones ===")
+    ok_json = ok_quality = 0
+    for goal, snippets in PLAN_CASES:
+        user = f"Goal: {goal}\n\nResearch snippets:\n" + "\n".join(f"- {s}" for s in snippets)
+        raw, parsed, secs = ask(llm, PLAN_SYS, user, PLAN_SCHEMA, max_tokens=640)
+        ms = (parsed or {}).get("milestones") if parsed else None
+        good = isinstance(ms, list) and 3 <= len(ms) <= 6 and all(
+            isinstance(m, dict) and m.get("title") for m in ms)
+        ok_json += parsed is not None
+        ok_quality += good
+        print(f"  [{secs:4.1f}s] json={'Y' if parsed else 'N'} 3-6ms={'Y' if good else 'N'} | {goal}")
+        if isinstance(ms, list):
+            for m in ms[:6]:
+                if isinstance(m, dict):
+                    print(f"        - {str(m.get('title'))[:60]}")
+    print(f"  -> valid-JSON {ok_json}/{len(PLAN_CASES)} | 3-6 milestones {ok_quality}/{len(PLAN_CASES)}\n")
+
+
+def run_quest(llm):
+    print("=== TASK 4: milestone -> themed quest title + narrative ===")
+    ok_json = ok_quality = 0
+    for task, theme in QUEST_CASES:
+        raw, parsed, secs = ask(llm, QUEST_SYS, f"Theme: {theme}\nTask: {task}", QUEST_SCHEMA, max_tokens=256)
+        good = bool(parsed) and bool((parsed or {}).get("quest_title")) and bool((parsed or {}).get("narrative"))
+        ok_json += parsed is not None
+        ok_quality += good
+        title = (parsed or {}).get("quest_title", "")
+        narr = (parsed or {}).get("narrative", "")
+        print(f"  [{secs:4.1f}s] json={'Y' if parsed else 'N'} | ({theme}) {task[:34]:34} -> {title[:40]}")
+        if narr:
+            print(f"        {narr[:96]}")
+    print(f"  -> valid-JSON {ok_json}/{len(QUEST_CASES)} | usable {ok_quality}/{len(QUEST_CASES)}\n")
+
+
+def main():
+    llm = load_llm()
+    run_intent(llm)
+    run_ask(llm)
+    run_plan(llm)
+    run_quest(llm)
+    print("[probe] done. Read the outputs: valid-JSON rate is the floor; the text quality decides\n"
+          "        which tasks need a heavier code fallback in the real build.")
+
+
+if __name__ == "__main__":
+    main()

store.py

@@ -0,0 +1,112 @@
+"""Durable per-user storage for FrogQuest — a single SQLite DB holding the WHOLE app state.
+
+Why this exists: `gr.BrowserState` (localStorage) is wiped by OS/browser cache clears and capped
+at ~5MB, so quests/campaigns/images don't survive on real devices. With HF login giving a stable
+per-user id, we persist the entire state dict (profile, campaigns, quests, generated images, chat,
+pending) server-side, keyed by that id. BrowserState stays as a fast cache / anonymous fallback;
+SQLite is the durable source of truth.
+
+Storage location:
+  - On a HF Space with Persistent Storage, `/data` is mounted and survives restarts -> `/data/frogquest.db`.
+  - Locally (dev) `/data` doesn't exist -> fall back to `./.frogquest/frogquest.db`.
+  - `FROGQUEST_DB_PATH` overrides both.
+
+Schema is deliberately trivial: one row per user, the state stored as a JSON blob (mirrors the
+app's in-memory state dict exactly, so there's no object-relational mapping to drift). Images are
+data-URL strings inside that JSON — SQLite has no 5MB cap, so `_trim_images` is unnecessary here.
+"""
+from __future__ import annotations
+
+import json
+import os
+import sqlite3
+import threading
+from datetime import datetime, timezone
+
+_lock = threading.Lock()
+_conn: sqlite3.Connection | None = None
+
+
+def _db_path() -> str:
+    override = os.environ.get("FROGQUEST_DB_PATH")
+    if override:
+        return override
+    if os.path.isdir("/data") and os.access("/data", os.W_OK):  # HF Persistent Storage
+        return "/data/frogquest.db"
+    base = os.path.join(os.path.dirname(os.path.abspath(__file__)), ".frogquest")  # local dev
+    os.makedirs(base, exist_ok=True)
+    return os.path.join(base, "frogquest.db")
+
+
+def _get_conn() -> sqlite3.Connection:
+    """Lazily open one shared connection. check_same_thread=False + the module _lock make it safe
+    across Gradio's worker threads; WAL keeps reads from blocking the single writer."""
+    global _conn
+    if _conn is None:
+        path = _db_path()
+        parent = os.path.dirname(path)
+        if parent:
+            os.makedirs(parent, exist_ok=True)
+        _conn = sqlite3.connect(path, check_same_thread=False)
+        _conn.execute("PRAGMA journal_mode=WAL")
+        _conn.execute("PRAGMA synchronous=NORMAL")
+        _conn.execute(
+            "CREATE TABLE IF NOT EXISTS users ("
+            "  uid TEXT PRIMARY KEY,"
+            "  state TEXT NOT NULL,"
+            "  updated_at TEXT NOT NULL"
+            ")"
+        )
+        _conn.commit()
+    return _conn
+
+
+def load(uid: str | None) -> dict | None:
+    """Return the saved state dict for this user, or None if absent / not logged in / unreadable."""
+    if not uid:
+        return None
+    with _lock:
+        row = _get_conn().execute("SELECT state FROM users WHERE uid = ?", (uid,)).fetchone()
+    if not row:
+        return None
+    try:
+        data = json.loads(row[0])
+        return data if isinstance(data, dict) else None
+    except (json.JSONDecodeError, TypeError):
+        return None
+
+
+def save(uid: str | None, state: dict) -> None:
+    """Upsert the full state dict for this user. No-op when not logged in (anonymous -> BrowserState
+    only). Never raises into the request path: persistence failures are swallowed and logged."""
+    if not uid or not isinstance(state, dict):
+        return
+    try:
+        blob = json.dumps(state, ensure_ascii=False)
+    except (TypeError, ValueError):
+        return
+    now = datetime.now(timezone.utc).isoformat()
+    try:
+        with _lock:
+            conn = _get_conn()
+            conn.execute(
+                "INSERT INTO users (uid, state, updated_at) VALUES (?, ?, ?) "
+                "ON CONFLICT(uid) DO UPDATE SET state=excluded.state, updated_at=excluded.updated_at",
+                (uid, blob, now),
+            )
+            conn.commit()
+    except sqlite3.Error as e:
+        print(f"[store] save failed for uid={uid!r}: {e}")
+
+
+if __name__ == "__main__":  # quick self-test (no model, no network)
+    os.environ.setdefault("FROGQUEST_DB_PATH", os.path.join(os.path.dirname(__file__), ".frogquest", "selftest.db"))
+    assert load("nobody") is None
+    sample = {"profile": {"theme": "fantasy"}, "quests": [{"id": "q1"}], "images": {"q1": {"initial": "data:..."}}}
+    save("user-123", sample)
+    got = load("user-123")
+    assert got == sample, got
+    sample["quests"].append({"id": "q2"})
+    save("user-123", sample)
+    assert len(load("user-123")["quests"]) == 2
+    print("store.py self-test OK ->", _db_path())