feat: cache dashboard quest analysis

Co-authored-by: Codex <noreply@openai.com>

README.md

@@ -86,9 +86,20 @@ then swaps the live app to the new dashboard. `GET /api/dashboard/refresh` polls
 
 Live refresh requires a writable dashboard cache directory at `ADVISOR_CACHE_DIR`. On Hugging Face Spaces this should be
 a mounted Storage Bucket; locally it can be a normal directory such as `.cache/advisor-dashboard`. The job writes
-`runs/{run_id}/projects.json`, `project_index.json`, `dashboard.json`, and `manifest.json`, then atomically updates
-`latest.json`. If the cache directory is missing, not writable, or quest analysis fails validation, refresh fails and the
-current validated dashboard stays active.
+`runs/{run_id}/projects.json`, `project_index.json`, `dashboard.json`, `quest_analysis.json`, and `manifest.json`, then
+atomically updates `latest.json`. Quest analysis also keeps validated per-project records under
+`quest-cache/v1/{prefix}/{cache_key}.json`, keyed by the rendered README+app-file prompt hash, taxonomy hash, MiniCPM
+model id, adapter id/revision, local adapter digest, and generation config. Refresh logs every cache hit, miss, and newly
+analyzed project. If the cache directory is missing, not writable, or quest analysis fails validation, refresh fails and
+the current validated dashboard stays active.
+
+When `ADVISOR_CACHE_DIR` is set, the app starts a scheduler thread that checks once per hour and starts a normal
+dashboard refresh if no refresh is already running. `ADVISOR_SCHEDULED_REFRESH=0` or
+`ADVISOR_DISABLE_SCHEDULED_REFRESH=1` disables it; `ADVISOR_REFRESH_INTERVAL_SECONDS`,
+`ADVISOR_REFRESH_INITIAL_DELAY_SECONDS`, and `ADVISOR_SCHEDULED_REFRESH_COMPUTE` tune the cadence and compute mode.
+Manual and scheduled refreshes both acquire `$ADVISOR_CACHE_DIR/refresh.lock` atomically before work starts, so multiple
+app processes do not analyze the same snapshot concurrently. Stale locks expire after `ADVISOR_REFRESH_LOCK_TTL_SECONDS`
+(default two hours).
 
 Set `ADVISOR_QUEST_ANALYZER_BACKEND=minicpm-transformers` for both local and deployed refresh runs. The local dashboard
 uses the same MiniCPM analyzer as the deployed Space; test doubles are only used inside pytest.
@@ -221,6 +232,11 @@ ADVISOR_QUEST_ANALYZER_BACKEND=minicpm-transformers
 ADVISOR_QUEST_ADAPTER_ID=artifacts/quest-lora
 ADVISOR_QUEST_ANALYSIS_BATCH_SIZE=8
 ADVISOR_CACHE_DIR=/data/advisor-cache
+ADVISOR_REFRESH_COMPUTE=cpu
+ADVISOR_SCHEDULED_REFRESH_COMPUTE=cpu
+ADVISOR_REFRESH_INTERVAL_SECONDS=3600
+ADVISOR_REFRESH_INITIAL_DELAY_SECONDS=300
+ADVISOR_REFRESH_LOCK_TTL_SECONDS=7200
 ADVISOR_REFRESH_EMBEDDING_TIMEOUT_SECONDS=1800
 ADVISOR_EMBEDDING_MODEL_REPO=ggml-org/embeddinggemma-300m-qat-q8_0-GGUF
 ADVISOR_EMBEDDING_MODEL_FILE=embeddinggemma-300m-qat-Q8_0.gguf

app.py

@@ -49,6 +49,13 @@ from hackathon_advisor.lora_dataset import build_lora_dataset_jsonl
 from hackathon_advisor.lora_training_kit import TRAINING_KIT_FILENAME, build_lora_training_kit_zip
 from hackathon_advisor.png_export import artifact_png_filename, render_artifact_png
 from hackathon_advisor.prize_ledger import prize_ledger
+from hackathon_advisor.quest_cache import (
+    build_quest_analysis_run_payload,
+    quest_analyzer_fingerprint_from_env,
+    quest_cache_run_record,
+    read_quest_cache_entry,
+    write_quest_cache_entry,
+)
 from hackathon_advisor.quest_analysis import create_quest_analyzer, validate_matches_by_project
 from hackathon_advisor.runtime_hooks import install_asyncio_cleanup_hook
 from hackathon_advisor.submission_packet import build_submission_packet_markdown
@@ -71,6 +78,11 @@ AUDIO_UPLOAD_SUFFIXES = {".aac", ".aif", ".aiff", ".flac", ".m4a", ".mp3", ".oga
 DEFAULT_HF_ORG = "build-small-hackathon"
 DEFAULT_REFRESH_EMBEDDING_TIMEOUT_SECONDS = 1800
 DEFAULT_QUEST_ANALYSIS_BATCH_SIZE = 8
+DEFAULT_REFRESH_COMPUTE = "cpu"
+DEFAULT_SCHEDULED_REFRESH_INTERVAL_SECONDS = 3600
+DEFAULT_SCHEDULED_REFRESH_INITIAL_DELAY_SECONDS = 300
+DEFAULT_REFRESH_LOCK_TTL_SECONDS = 7200
+REFRESH_LOCK_FILENAME = "refresh.lock"
 REFRESH_SUBPROCESS_LOG_TAIL_LINES = 80
 REFRESH_STAGE_LABELS = {
     "crawling": "Fetching public Spaces",
@@ -83,6 +95,19 @@ REFRESH_STAGE_LABELS = {
 
 _runtime_lock = Lock()
 _refresh_lock = Lock()
+_scheduler_lock = Lock()
+_scheduler_started = False
+
+
+def _empty_quest_cache_progress() -> dict[str, Any]:
+    return {
+        "project_count": 0,
+        "hit_count": 0,
+        "miss_count": 0,
+        "analyzed_count": 0,
+        "remaining_count": 0,
+        "last_project_id": "",
+    }
 
 
 def _load_initial_runtime() -> tuple[ProjectIndex, dict[str, Any]]:
@@ -107,12 +132,15 @@ _cpu_engine: AdvisorEngine | None = None
 _refresh_state: dict[str, Any] = {
     "status": "idle",
     "run_id": "",
+    "compute": "",
+    "reason": "",
     "stage": "",
     "stage_label": "",
     "started_at": "",
     "finished_at": "",
     "error": "",
     "result": None,
+    "quest_cache": _empty_quest_cache_progress(),
 }
 
 
@@ -140,7 +168,13 @@ def _transcribe_voice(audio_path: str) -> dict[str, Any]:
     return voice_transcriber.transcribe(Path(audio_path)).to_dict()
 
 
-def _analyze_dashboard_quests(project_rows: list[dict[str, Any]]) -> dict[str, Any]:
+def _analyze_dashboard_quests(
+    project_rows: list[dict[str, Any]],
+    *,
+    cache_dir: Path,
+    compute: str,
+    run_id: str,
+) -> dict[str, Any]:
     missing_evidence_keys = [
         str(item.get("id") or index)
         for index, item in enumerate(project_rows)
@@ -152,25 +186,144 @@ def _analyze_dashboard_quests(project_rows: list[dict[str, Any]]) -> dict[str, A
             f"missing evidence keys for {len(missing_evidence_keys)} projects"
         )
     projects = [Project.from_dict(item) for item in project_rows]
+    analyzer_fingerprint = quest_analyzer_fingerprint_from_env()
     matches_by_project: dict[str, list[dict[str, Any]]] = {}
-    source = "quest-analyzer"
+    record_by_project: dict[str, dict[str, Any]] = {}
+    misses: list[tuple[Project, dict[str, Any]]] = []
+    hit_count = 0
+    miss_count = 0
+    analyzed_count = 0
+    source = str(analyzer_fingerprint["source"])
     batch_size = _quest_analysis_batch_size()
-    for start in range(0, len(project_rows), batch_size):
-        batch_rows = project_rows[start : start + batch_size]
-        result = _analyze_dashboard_quest_batch(batch_rows)
+    _set_quest_cache_progress(
+        project_count=len(projects),
+        hit_count=0,
+        miss_count=0,
+        analyzed_count=0,
+        remaining_count=len(projects),
+        last_project_id="",
+    )
+
+    for project in projects:
+        lookup = read_quest_cache_entry(cache_dir, project, analyzer_fingerprint)
+        if lookup.entry is not None:
+            hit_count += 1
+            matches_by_project[project.id] = lookup.entry.matches
+            record_by_project[project.id] = quest_cache_run_record(
+                project=project,
+                identity=lookup.identity,
+                matches=lookup.entry.matches,
+                status="cached",
+                source=lookup.entry.source,
+                path=lookup.entry.path,
+            )
+            print(
+                f"[quest-cache] hit {project.id} key={lookup.identity.cache_key[:12]} "
+                f"matches={len(lookup.entry.matches)}",
+                flush=True,
+            )
+        else:
+            miss_count += 1
+            misses.append((project, lookup.identity.to_dict()))
+            print(
+                f"[quest-cache] miss {project.id} key={lookup.identity.cache_key[:12]} "
+                f"reason={lookup.reason}",
+                flush=True,
+            )
+        _set_quest_cache_progress(
+            project_count=len(projects),
+            hit_count=hit_count,
+            miss_count=miss_count,
+            analyzed_count=analyzed_count,
+            remaining_count=len(projects) - hit_count - analyzed_count,
+            last_project_id=project.id,
+        )
+
+    for start in range(0, len(misses), batch_size):
+        batch = misses[start : start + batch_size]
+        batch_projects = [item[0] for item in batch]
+        batch_rows = [project.to_refresh_snapshot_dict() for project in batch_projects]
+        result = _analyze_dashboard_quest_batch(batch_rows, compute=compute)
         source = str(result["source"])
-        matches_by_project.update(result["matches_by_project"])
+        validated_batch = validate_matches_by_project(
+            result["matches_by_project"],
+            batch_projects,
+            source=source,
+        )
+        for project, _identity_row in batch:
+            entry = write_quest_cache_entry(
+                cache_dir,
+                project,
+                analyzer_fingerprint,
+                validated_batch.matches_by_project[project.id],
+                source=source,
+            )
+            analyzed_count += 1
+            matches_by_project[project.id] = entry.matches
+            record_by_project[project.id] = quest_cache_run_record(
+                project=project,
+                identity=entry.identity,
+                matches=entry.matches,
+                status="analyzed",
+                source=entry.source,
+                path=entry.path,
+            )
+            print(
+                f"[quest-cache] analyzed {project.id} key={entry.identity.cache_key[:12]} "
+                f"matches={len(entry.matches)}",
+                flush=True,
+            )
+            _set_quest_cache_progress(
+                project_count=len(projects),
+                hit_count=hit_count,
+                miss_count=miss_count,
+                analyzed_count=analyzed_count,
+                remaining_count=len(projects) - hit_count - analyzed_count,
+                last_project_id=project.id,
+            )
     validated = validate_matches_by_project(matches_by_project, projects, source=source)
+    summary = {
+        "project_count": len(projects),
+        "hit_count": hit_count,
+        "miss_count": miss_count,
+        "analyzed_count": analyzed_count,
+        "remaining_count": 0,
+        "compute": compute,
+    }
+    project_records = [record_by_project[project.id] for project in projects]
     return {
         "source": validated.source,
         "matches_by_project": validated.matches_by_project,
+        "quest_analysis_payload": build_quest_analysis_run_payload(
+            run_id=run_id,
+            analyzer_fingerprint=analyzer_fingerprint,
+            summary=summary,
+            project_records=project_records,
+        ),
     }
 
 
 @gpu_task
-def _analyze_dashboard_quest_batch(project_rows: list[dict[str, Any]]) -> dict[str, Any]:
+def _analyze_dashboard_quest_batch_gpu(project_rows: list[dict[str, Any]]) -> dict[str, Any]:
+    return _analyze_dashboard_quest_batch_with_device(
+        project_rows,
+        device="cuda" if zero_gpu_enabled() else "local",
+    )
+
+
+def _analyze_dashboard_quest_batch_cpu(project_rows: list[dict[str, Any]]) -> dict[str, Any]:
+    return _analyze_dashboard_quest_batch_with_device(project_rows, device="cpu")
+
+
+def _analyze_dashboard_quest_batch(project_rows: list[dict[str, Any]], *, compute: str) -> dict[str, Any]:
+    if compute == "gpu":
+        return _analyze_dashboard_quest_batch_gpu(project_rows)
+    return _analyze_dashboard_quest_batch_cpu(project_rows)
+
+
+def _analyze_dashboard_quest_batch_with_device(project_rows: list[dict[str, Any]], *, device: str) -> dict[str, Any]:
     projects = [Project.from_dict(item) for item in project_rows]
-    analyzer = create_quest_analyzer(device="cuda" if zero_gpu_enabled() else "local")
+    analyzer = create_quest_analyzer(device=device)
     matches = analyzer.analyze(projects)
     source = getattr(analyzer, "source", "quest-analyzer")
     validated = validate_matches_by_project(matches, projects, source=source)
@@ -192,41 +345,191 @@ def _quest_analysis_batch_size() -> int:
 
 def _refresh_public_state() -> dict[str, Any]:
     with _refresh_lock:
-        return dict(_refresh_state)
+        state = dict(_refresh_state)
+        state["quest_cache"] = dict(_refresh_state.get("quest_cache") or _empty_quest_cache_progress())
+        return state
 
 
 def _set_refresh_state(**updates: Any) -> None:
     with _refresh_lock:
+        if "quest_cache" in updates:
+            updates["quest_cache"] = dict(updates["quest_cache"])
         _refresh_state.update(updates)
         stage = str(_refresh_state.get("stage") or "")
         _refresh_state["stage_label"] = REFRESH_STAGE_LABELS.get(stage, "")
 
 
-def _start_refresh_thread(cache_dir: Path) -> dict[str, Any]:
+def _set_quest_cache_progress(**updates: Any) -> None:
+    with _refresh_lock:
+        progress = dict(_refresh_state.get("quest_cache") or _empty_quest_cache_progress())
+        progress.update(updates)
+        _refresh_state["quest_cache"] = progress
+
+
+def _normalize_refresh_compute(value: Any) -> str:
+    compute = str(value or "").strip().lower() or DEFAULT_REFRESH_COMPUTE
+    if compute not in {"cpu", "gpu"}:
+        raise HTTPException(status_code=400, detail="Dashboard refresh compute must be 'cpu' or 'gpu'.")
+    return compute
+
+
+def _default_refresh_compute() -> str:
+    return _normalize_refresh_compute(os.environ.get("ADVISOR_REFRESH_COMPUTE", DEFAULT_REFRESH_COMPUTE))
+
+
+def _refresh_lock_ttl_seconds() -> int:
+    raw = os.environ.get("ADVISOR_REFRESH_LOCK_TTL_SECONDS", "").strip()
+    if not raw:
+        return DEFAULT_REFRESH_LOCK_TTL_SECONDS
+    ttl = int(raw)
+    if ttl <= 0:
+        raise RuntimeError("ADVISOR_REFRESH_LOCK_TTL_SECONDS must be a positive integer.")
+    return ttl
+
+
+def _refresh_lock_path(cache_dir: Path) -> Path:
+    return cache_dir / REFRESH_LOCK_FILENAME
+
+
+def _acquire_refresh_lease(cache_dir: Path, *, run_id: str, compute: str, reason: str) -> None:
+    lock_path = _refresh_lock_path(cache_dir)
+    now = time.time()
+    lease = {
+        "schema_version": 1,
+        "run_id": run_id,
+        "compute": compute,
+        "reason": reason,
+        "owner": _refresh_owner(),
+        "started_at": datetime.now(timezone.utc).isoformat(timespec="seconds"),
+        "expires_at_epoch": now + _refresh_lock_ttl_seconds(),
+    }
+    while True:
+        try:
+            fd = os.open(lock_path, os.O_CREAT | os.O_EXCL | os.O_WRONLY, 0o644)
+        except FileExistsError as error:
+            existing = _read_refresh_lease(lock_path)
+            if existing is None or _refresh_lease_expired(existing):
+                run_label = str((existing or {}).get("run_id") or "unknown")
+                print(f"[dashboard-refresh] removing stale refresh lock run={run_label}", flush=True)
+                try:
+                    lock_path.unlink()
+                except FileNotFoundError:
+                    pass
+                except OSError as unlink_error:
+                    raise HTTPException(
+                        status_code=409,
+                        detail=f"Dashboard refresh lock exists and could not be removed: {unlink_error}",
+                    ) from unlink_error
+                continue
+            raise HTTPException(
+                status_code=409,
+                detail=(
+                    "Dashboard refresh is already running "
+                    f"(run {existing.get('run_id', 'unknown')}, owner {existing.get('owner', 'unknown')})."
+                ),
+            ) from error
+        with os.fdopen(fd, "w", encoding="utf-8") as handle:
+            handle.write(json.dumps(lease, ensure_ascii=False) + "\n")
+        print(
+            f"[dashboard-refresh] acquired refresh lock run={run_id} compute={compute} reason={reason}",
+            flush=True,
+        )
+        return
+
+
+def _release_refresh_lease(cache_dir: Path, run_id: str) -> None:
+    lock_path = _refresh_lock_path(cache_dir)
+    existing = _read_refresh_lease(lock_path)
+    if existing is None:
+        return
+    if str(existing.get("run_id") or "") != run_id:
+        print(
+            f"[dashboard-refresh] refresh lock belongs to {existing.get('run_id', 'unknown')}; "
+            f"not releasing run={run_id}",
+            flush=True,
+        )
+        return
+    try:
+        lock_path.unlink()
+    except FileNotFoundError:
+        return
+    print(f"[dashboard-refresh] released refresh lock run={run_id}", flush=True)
+
+
+def _read_refresh_lease(lock_path: Path) -> dict[str, Any] | None:
+    try:
+        payload = json.loads(lock_path.read_text(encoding="utf-8"))
+    except FileNotFoundError:
+        return None
+    except (OSError, json.JSONDecodeError):
+        return None
+    return payload if isinstance(payload, dict) else None
+
+
+def _refresh_lease_expired(lease: dict[str, Any]) -> bool:
+    try:
+        expires_at = float(lease.get("expires_at_epoch"))
+    except (TypeError, ValueError):
+        return True
+    return expires_at <= time.time()
+
+
+def _refresh_owner() -> str:
+    node = getattr(os, "uname", lambda: None)()
+    host = getattr(node, "nodename", "") if node is not None else ""
+    return f"{host or 'process'}:{os.getpid()}"
+
+
+def _start_refresh_thread(cache_dir: Path, *, compute: str, reason: str) -> dict[str, Any]:
+    compute = _normalize_refresh_compute(compute)
     with _refresh_lock:
         if _refresh_state.get("status") == "running":
             raise HTTPException(status_code=409, detail="Dashboard refresh is already running.")
         run_id = datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%SZ") + "-" + uuid4().hex[:8]
+        _acquire_refresh_lease(cache_dir, run_id=run_id, compute=compute, reason=reason)
         _refresh_state.update(
             {
                 "status": "running",
                 "run_id": run_id,
+                "compute": compute,
+                "reason": reason,
                 "stage": "crawling",
                 "stage_label": REFRESH_STAGE_LABELS["crawling"],
                 "started_at": datetime.now(timezone.utc).isoformat(timespec="seconds"),
                 "finished_at": "",
                 "error": "",
                 "result": None,
+                "quest_cache": _empty_quest_cache_progress(),
             }
         )
-    thread = Thread(target=_run_refresh_job, args=(run_id, cache_dir), daemon=True)
-    thread.start()
+    thread = Thread(target=_run_refresh_job, args=(run_id, cache_dir, compute), daemon=True)
+    try:
+        thread.start()
+    except Exception:
+        _release_refresh_lease(cache_dir, run_id)
+        _set_refresh_state(
+            status="idle",
+            run_id="",
+            compute="",
+            reason="",
+            stage="",
+            started_at="",
+            finished_at="",
+            error="",
+            result=None,
+            quest_cache=_empty_quest_cache_progress(),
+        )
+        raise
     return _refresh_public_state()
 
 
-def _run_refresh_job(run_id: str, cache_dir: Path) -> None:
+def _run_refresh_job(run_id: str, cache_dir: Path, compute: str) -> None:
     try:
-        projects_payload, index_payload, refreshed_dashboard = _build_refresh_payloads(run_id)
+        projects_payload, index_payload, refreshed_dashboard, quest_analysis_payload = _build_refresh_payloads(
+            run_id,
+            cache_dir=cache_dir,
+            compute=compute,
+        )
         _set_refresh_state(stage="persisting")
         artifacts = persist_refresh_artifacts(
             cache_dir,
@@ -234,9 +537,11 @@ def _run_refresh_job(run_id: str, cache_dir: Path) -> None:
             projects_payload=projects_payload,
             index_payload=index_payload,
             dashboard_payload=refreshed_dashboard,
+            quest_analysis_payload=quest_analysis_payload,
         )
         _set_refresh_state(stage="swapping")
         _replace_runtime_from_files(artifacts.projects_path, artifacts.index_path, artifacts.dashboard)
+        _release_refresh_lease(cache_dir, run_id)
         _set_refresh_state(
             status="succeeded",
             stage="",
@@ -246,11 +551,13 @@ def _run_refresh_job(run_id: str, cache_dir: Path) -> None:
                 "project_count": refreshed_dashboard["project_count"],
                 "snapshot_digest": refreshed_dashboard["provenance"]["snapshot_digest"],
                 "dashboard_generated_at": refreshed_dashboard["generated_at"],
+                "quest_cache": dict(quest_analysis_payload.get("summary") or {}),
             },
         )
     except Exception as error:  # noqa: BLE001 - background job must report every failure as state
         print("[dashboard-refresh] failed", flush=True)
         traceback.print_exception(type(error), error, error.__traceback__)
+        _release_refresh_lease(cache_dir, run_id)
         _set_refresh_state(
             status="failed",
             stage="",
@@ -258,9 +565,16 @@ def _run_refresh_job(run_id: str, cache_dir: Path) -> None:
             error=_format_refresh_error(error),
             result=None,
         )
+    finally:
+        _release_refresh_lease(cache_dir, run_id)
 
 
-def _build_refresh_payloads(run_id: str) -> tuple[dict[str, Any], dict[str, Any], dict[str, Any]]:
+def _build_refresh_payloads(
+    run_id: str,
+    *,
+    cache_dir: Path,
+    compute: str,
+) -> tuple[dict[str, Any], dict[str, Any], dict[str, Any], dict[str, Any]]:
     from scripts.crawl_hf_spaces import API, crawl_projects
 
     org = os.environ.get("ADVISOR_HF_ORG", DEFAULT_HF_ORG).strip() or DEFAULT_HF_ORG
@@ -294,14 +608,19 @@ def _build_refresh_payloads(run_id: str) -> tuple[dict[str, Any], dict[str, Any]
     )
 
     _set_refresh_state(stage="quest_analysis")
-    quest_analysis = _analyze_dashboard_quests([project.to_refresh_snapshot_dict() for project in projects])
+    quest_analysis = _analyze_dashboard_quests(
+        [project.to_refresh_snapshot_dict() for project in projects],
+        cache_dir=cache_dir,
+        compute=compute,
+        run_id=run_id,
+    )
     _set_refresh_state(stage="atlas")
     refreshed_dashboard = build_dashboard_payload(
         refreshed_index,
         quest_matches=quest_analysis["matches_by_project"],
         quest_source=str(quest_analysis["source"]),
     )
-    return projects_payload, index_payload, refreshed_dashboard
+    return projects_payload, index_payload, refreshed_dashboard, quest_analysis["quest_analysis_payload"]
 
 
 def _build_refresh_index_payload(
@@ -561,12 +880,13 @@ def dashboard() -> dict:
 
 
 @app.post("/api/dashboard/refresh")
-def dashboard_refresh_start() -> JSONResponse:
+def dashboard_refresh_start(payload: dict[str, Any] | None = None) -> JSONResponse:
     try:
         cache_dir = require_writable_cache_dir()
     except DashboardStorageError as error:
         raise HTTPException(status_code=400, detail=str(error)) from error
-    return JSONResponse(_start_refresh_thread(cache_dir), status_code=202)
+    compute = _refresh_compute_from_payload(payload)
+    return JSONResponse(_start_refresh_thread(cache_dir, compute=compute, reason="manual"), status_code=202)
 
 
 @app.get("/api/dashboard/refresh")
@@ -574,6 +894,104 @@ def dashboard_refresh_status() -> dict:
     return _refresh_public_state()
 
 
+def _refresh_compute_from_payload(payload: dict[str, Any] | None) -> str:
+    payload = payload or {}
+    return _normalize_refresh_compute(payload.get("compute") or _default_refresh_compute())
+
+
+def _start_scheduled_refresh_loop() -> None:
+    global _scheduler_started
+    if not _scheduled_refresh_enabled():
+        return
+    with _scheduler_lock:
+        if _scheduler_started:
+            return
+        _scheduler_started = True
+    interval = _scheduled_refresh_interval_seconds()
+    initial_delay = _scheduled_refresh_initial_delay_seconds()
+    compute = _scheduled_refresh_compute()
+    print(
+        "[dashboard-refresh scheduler] enabled "
+        f"interval={interval}s initial_delay={initial_delay}s compute={compute}",
+        flush=True,
+    )
+    Thread(
+        target=_scheduled_refresh_loop,
+        args=(interval, initial_delay),
+        daemon=True,
+        name="dashboard-refresh-scheduler",
+    ).start()
+
+
+def _scheduled_refresh_enabled() -> bool:
+    disabled = os.environ.get("ADVISOR_DISABLE_SCHEDULED_REFRESH", "").strip().lower()
+    if disabled in {"1", "true", "yes", "on"}:
+        return False
+    raw = os.environ.get("ADVISOR_SCHEDULED_REFRESH", "").strip().lower()
+    if raw:
+        return raw in {"1", "true", "yes", "on"}
+    return cache_dir_from_env() is not None
+
+
+def _scheduled_refresh_interval_seconds() -> int:
+    raw = (
+        os.environ.get("ADVISOR_REFRESH_INTERVAL_SECONDS", "").strip()
+        or os.environ.get("ADVISOR_SCHEDULED_REFRESH_INTERVAL_SECONDS", "").strip()
+    )
+    if not raw:
+        return DEFAULT_SCHEDULED_REFRESH_INTERVAL_SECONDS
+    interval = int(raw)
+    if interval <= 0:
+        raise RuntimeError("ADVISOR_REFRESH_INTERVAL_SECONDS must be a positive integer.")
+    return interval
+
+
+def _scheduled_refresh_initial_delay_seconds() -> int:
+    raw = os.environ.get("ADVISOR_REFRESH_INITIAL_DELAY_SECONDS", "").strip()
+    if not raw:
+        return DEFAULT_SCHEDULED_REFRESH_INITIAL_DELAY_SECONDS
+    delay = int(raw)
+    if delay < 0:
+        raise RuntimeError("ADVISOR_REFRESH_INITIAL_DELAY_SECONDS must not be negative.")
+    return delay
+
+
+def _scheduled_refresh_compute() -> str:
+    return _normalize_refresh_compute(
+        os.environ.get("ADVISOR_SCHEDULED_REFRESH_COMPUTE", "").strip() or _default_refresh_compute()
+    )
+
+
+def _scheduled_refresh_loop(interval_seconds: int, initial_delay_seconds: int) -> None:
+    if initial_delay_seconds:
+        time.sleep(initial_delay_seconds)
+    while True:
+        _run_scheduled_refresh_once()
+        time.sleep(interval_seconds)
+
+
+def _run_scheduled_refresh_once() -> None:
+    try:
+        cache_dir = require_writable_cache_dir()
+        state = _start_refresh_thread(
+            cache_dir,
+            compute=_scheduled_refresh_compute(),
+            reason="scheduled",
+        )
+        print(
+            f"[dashboard-refresh scheduler] started run={state.get('run_id', '')} "
+            f"compute={state.get('compute', '')}",
+            flush=True,
+        )
+    except HTTPException as error:
+        if error.status_code == 409:
+            print(f"[dashboard-refresh scheduler] skipped: {error.detail}", flush=True)
+            return
+        print(f"[dashboard-refresh scheduler] failed to start: {error.detail}", flush=True)
+    except Exception as error:  # noqa: BLE001 - scheduler must keep running after transient failures
+        print(f"[dashboard-refresh scheduler] failed to start: {_format_refresh_error(error)}", flush=True)
+
+
 @app.get("/health")
 def health() -> dict:
     return {
@@ -817,6 +1235,9 @@ def agent_turn(message: str, session_json: str = "{}", compute: str = "gpu") ->
     yield from _agent_turn_events(message, session_json, _normalize_compute(compute))
 
 
+_start_scheduled_refresh_loop()
+
+
 if __name__ == "__main__":
     app.launch(
         server_name=os.environ.get("GRADIO_SERVER_NAME", "0.0.0.0"),

hackathon_advisor/dashboard_storage.py

@@ -27,6 +27,7 @@ class DashboardArtifacts:
     manifest_path: Path
     dashboard: dict[str, Any]
     manifest: dict[str, Any]
+    quest_analysis_path: Path | None = None
 
 
 def cache_dir_from_env(env: dict[str, str] | None = None) -> Path | None:
@@ -88,6 +89,7 @@ def persist_refresh_artifacts(
     projects_payload: dict[str, Any],
     index_payload: dict[str, Any],
     dashboard_payload: dict[str, Any],
+    quest_analysis_payload: dict[str, Any] | None = None,
 ) -> DashboardArtifacts:
     validate_dashboard_payload(dashboard_payload)
     relative_run_dir = Path("runs") / run_id
@@ -97,21 +99,27 @@ def persist_refresh_artifacts(
     projects_path = run_dir / "projects.json"
     index_path = run_dir / "project_index.json"
     dashboard_path = run_dir / "dashboard.json"
+    quest_analysis_path = run_dir / "quest_analysis.json" if quest_analysis_payload is not None else None
     manifest_path = run_dir / "manifest.json"
     _write_json(projects_path, projects_payload)
     _write_json(index_path, index_payload)
     _write_json(dashboard_path, dashboard_payload)
+    if quest_analysis_path is not None:
+        _write_json(quest_analysis_path, quest_analysis_payload)
+    artifact_paths = {
+        "projects": _relative(cache_dir, projects_path),
+        "index": _relative(cache_dir, index_path),
+        "dashboard": _relative(cache_dir, dashboard_path),
+    }
+    if quest_analysis_path is not None:
+        artifact_paths["quest_analysis"] = _relative(cache_dir, quest_analysis_path)
     manifest = {
         "schema_version": STORAGE_SCHEMA_VERSION,
         "run_id": run_id,
         "generated_at": datetime.now(timezone.utc).isoformat(timespec="seconds"),
         "project_count": dashboard_payload["project_count"],
         "snapshot_digest": dashboard_payload["provenance"]["snapshot_digest"],
-        "artifacts": {
-            "projects": _relative(cache_dir, projects_path),
-            "index": _relative(cache_dir, index_path),
-            "dashboard": _relative(cache_dir, dashboard_path),
-        },
+        "artifacts": artifact_paths,
     }
     _write_json(manifest_path, manifest)
 
@@ -124,6 +132,8 @@ def persist_refresh_artifacts(
         "dashboard": _relative(cache_dir, dashboard_path),
         "manifest": _relative(cache_dir, manifest_path),
     }
+    if quest_analysis_path is not None:
+        latest["quest_analysis"] = _relative(cache_dir, quest_analysis_path)
     latest_path = cache_dir / LATEST_FILENAME
     tmp_path = cache_dir / f".{LATEST_FILENAME}.{run_id}.tmp"
     _write_json(tmp_path, latest)
@@ -135,6 +145,7 @@ def persist_refresh_artifacts(
         manifest_path=manifest_path,
         dashboard=dashboard_payload,
         manifest=manifest,
+        quest_analysis_path=quest_analysis_path,
     )
 
 

hackathon_advisor/quest_cache.py

@@ -0,0 +1,284 @@
+from __future__ import annotations
+
+from collections.abc import Mapping, Sequence
+from dataclasses import dataclass
+from datetime import datetime, timezone
+from hashlib import sha256
+import json
+import os
+from pathlib import Path
+from typing import Any
+from uuid import uuid4
+
+from hackathon_advisor.data import Project
+from hackathon_advisor.model_runtime import DEFAULT_MODEL_ID
+from hackathon_advisor.quest_analysis import (
+    DEFAULT_QUEST_ADAPTER_ID,
+    DEFAULT_QUEST_ADAPTER_REVISION,
+    MAX_QUEST_TOKENS,
+    QuestAnalysisError,
+    render_project_quest_prompt,
+    validate_matches_by_project,
+)
+from hackathon_advisor.quest_taxonomy import (
+    APP_PROMPT_CHAR_LIMIT,
+    QUEST_PROFILES,
+    QUEST_SYSTEM_PROMPT,
+    README_PROMPT_CHAR_LIMIT,
+)
+
+
+QUEST_CACHE_SCHEMA_VERSION = 1
+QUEST_CACHE_ROOT = Path("quest-cache") / "v1"
+QUEST_PROMPT_VERSION = "quest-prompt-v1"
+QUEST_ANALYZER_SOURCE = "minicpm-json-quest-analyzer"
+QUEST_GENERATION_CONFIG = {
+    "enable_thinking": False,
+    "temperature": 0.0,
+    "do_sample": False,
+    "max_new_tokens": MAX_QUEST_TOKENS,
+}
+
+
+@dataclass(frozen=True)
+class QuestCacheIdentity:
+    project_id: str
+    prompt_hash: str
+    taxonomy_hash: str
+    analyzer_fingerprint: dict[str, Any]
+    cache_key: str
+
+    def to_dict(self) -> dict[str, Any]:
+        return {
+            "project_id": self.project_id,
+            "prompt_hash": self.prompt_hash,
+            "taxonomy_hash": self.taxonomy_hash,
+            "analyzer_fingerprint": self.analyzer_fingerprint,
+            "cache_key": self.cache_key,
+        }
+
+
+@dataclass(frozen=True)
+class QuestCacheEntry:
+    identity: QuestCacheIdentity
+    matches: list[dict[str, Any]]
+    source: str
+    path: Path
+    generated_at: str
+
+
+@dataclass(frozen=True)
+class QuestCacheLookup:
+    identity: QuestCacheIdentity
+    entry: QuestCacheEntry | None
+    reason: str
+
+
+def quest_analyzer_fingerprint_from_env(env: Mapping[str, str] | None = None) -> dict[str, Any]:
+    values = env or os.environ
+    model_id = _first_env(values, "ADVISOR_QUEST_MODEL_ID", "ADVISOR_MODEL_ID") or DEFAULT_MODEL_ID
+    adapter_id = values.get("ADVISOR_QUEST_ADAPTER_ID", DEFAULT_QUEST_ADAPTER_ID).strip()
+    adapter_revision = values.get("ADVISOR_QUEST_ADAPTER_REVISION", DEFAULT_QUEST_ADAPTER_REVISION).strip()
+    return {
+        "source": QUEST_ANALYZER_SOURCE,
+        "model_id": model_id,
+        "adapter_id": adapter_id,
+        "adapter_revision": adapter_revision,
+        "adapter_digest": _local_artifact_digest(adapter_id),
+        "prompt_version": QUEST_PROMPT_VERSION,
+        "generation": dict(QUEST_GENERATION_CONFIG),
+    }
+
+
+def quest_taxonomy_hash() -> str:
+    payload = {
+        "system_prompt": QUEST_SYSTEM_PROMPT,
+        "quest_profiles": list(QUEST_PROFILES),
+        "readme_prompt_char_limit": README_PROMPT_CHAR_LIMIT,
+        "app_prompt_char_limit": APP_PROMPT_CHAR_LIMIT,
+        "prompt_version": QUEST_PROMPT_VERSION,
+    }
+    return sha256(_canonical_json(payload).encode("utf-8")).hexdigest()
+
+
+def build_quest_cache_identity(
+    project: Project,
+    analyzer_fingerprint: Mapping[str, Any],
+) -> QuestCacheIdentity:
+    prompt_hash = sha256(render_project_quest_prompt(project).encode("utf-8")).hexdigest()
+    taxonomy_hash = quest_taxonomy_hash()
+    canonical_fingerprint = json.loads(_canonical_json(analyzer_fingerprint))
+    key_payload = {
+        "schema_version": QUEST_CACHE_SCHEMA_VERSION,
+        "project_id": project.id,
+        "prompt_hash": prompt_hash,
+        "taxonomy_hash": taxonomy_hash,
+        "analyzer_fingerprint": canonical_fingerprint,
+    }
+    cache_key = sha256(_canonical_json(key_payload).encode("utf-8")).hexdigest()
+    return QuestCacheIdentity(
+        project_id=project.id,
+        prompt_hash=prompt_hash,
+        taxonomy_hash=taxonomy_hash,
+        analyzer_fingerprint=canonical_fingerprint,
+        cache_key=cache_key,
+    )
+
+
+def quest_cache_path(cache_dir: Path, cache_key: str) -> Path:
+    return cache_dir / QUEST_CACHE_ROOT / cache_key[:2] / f"{cache_key}.json"
+
+
+def read_quest_cache_entry(
+    cache_dir: Path,
+    project: Project,
+    analyzer_fingerprint: Mapping[str, Any],
+) -> QuestCacheLookup:
+    identity = build_quest_cache_identity(project, analyzer_fingerprint)
+    path = quest_cache_path(cache_dir, identity.cache_key)
+    if not path.is_file():
+        return QuestCacheLookup(identity=identity, entry=None, reason="absent")
+    try:
+        payload = json.loads(path.read_text(encoding="utf-8"))
+    except (OSError, json.JSONDecodeError) as error:
+        return QuestCacheLookup(identity=identity, entry=None, reason=f"invalid_json:{error}")
+    if not isinstance(payload, dict):
+        return QuestCacheLookup(identity=identity, entry=None, reason="invalid_payload")
+    try:
+        entry = _validate_cache_payload(payload, project, identity, path)
+    except QuestAnalysisError as error:
+        return QuestCacheLookup(identity=identity, entry=None, reason=f"invalid_schema:{error}")
+    return QuestCacheLookup(identity=identity, entry=entry, reason="hit")
+
+
+def write_quest_cache_entry(
+    cache_dir: Path,
+    project: Project,
+    analyzer_fingerprint: Mapping[str, Any],
+    matches: Sequence[Mapping[str, Any]],
+    *,
+    source: str,
+) -> QuestCacheEntry:
+    identity = build_quest_cache_identity(project, analyzer_fingerprint)
+    validated = validate_matches_by_project({project.id: list(matches)}, [project], source=source)
+    generated_at = datetime.now(timezone.utc).isoformat(timespec="seconds")
+    payload = {
+        "schema_version": QUEST_CACHE_SCHEMA_VERSION,
+        "generated_at": generated_at,
+        "source": validated.source,
+        **identity.to_dict(),
+        "matches": validated.matches_by_project[project.id],
+    }
+    path = quest_cache_path(cache_dir, identity.cache_key)
+    path.parent.mkdir(parents=True, exist_ok=True)
+    tmp_path = path.parent / f".{path.name}.{uuid4().hex}.tmp"
+    tmp_path.write_text(json.dumps(payload, indent=2, ensure_ascii=False) + "\n", encoding="utf-8")
+    os.replace(tmp_path, path)
+    return QuestCacheEntry(
+        identity=identity,
+        matches=validated.matches_by_project[project.id],
+        source=validated.source,
+        path=path,
+        generated_at=generated_at,
+    )
+
+
+def quest_cache_run_record(
+    *,
+    project: Project,
+    identity: QuestCacheIdentity,
+    matches: Sequence[Mapping[str, Any]],
+    status: str,
+    source: str,
+    path: Path | None = None,
+) -> dict[str, Any]:
+    return {
+        "project_id": project.id,
+        "cache_key": identity.cache_key,
+        "prompt_hash": identity.prompt_hash,
+        "taxonomy_hash": identity.taxonomy_hash,
+        "status": status,
+        "source": source,
+        "cache_path": path.as_posix() if path is not None else "",
+        "matches": [dict(match) for match in matches],
+    }
+
+
+def build_quest_analysis_run_payload(
+    *,
+    run_id: str,
+    analyzer_fingerprint: Mapping[str, Any],
+    summary: Mapping[str, Any],
+    project_records: Sequence[Mapping[str, Any]],
+) -> dict[str, Any]:
+    return {
+        "schema_version": QUEST_CACHE_SCHEMA_VERSION,
+        "run_id": run_id,
+        "generated_at": datetime.now(timezone.utc).isoformat(timespec="seconds"),
+        "source": QUEST_ANALYZER_SOURCE,
+        "analyzer_fingerprint": json.loads(_canonical_json(analyzer_fingerprint)),
+        "taxonomy_hash": quest_taxonomy_hash(),
+        "summary": dict(summary),
+        "projects": [dict(record) for record in project_records],
+    }
+
+
+def _validate_cache_payload(
+    payload: Mapping[str, Any],
+    project: Project,
+    identity: QuestCacheIdentity,
+    path: Path,
+) -> QuestCacheEntry:
+    if payload.get("schema_version") != QUEST_CACHE_SCHEMA_VERSION:
+        raise QuestAnalysisError("unsupported quest cache schema")
+    for field, expected in identity.to_dict().items():
+        if payload.get(field) != expected:
+            raise QuestAnalysisError(f"cache {field} mismatch")
+    source = str(payload.get("source") or QUEST_ANALYZER_SOURCE)
+    validated = validate_matches_by_project({project.id: payload.get("matches") or []}, [project], source=source)
+    generated_at = str(payload.get("generated_at") or "")
+    return QuestCacheEntry(
+        identity=identity,
+        matches=validated.matches_by_project[project.id],
+        source=validated.source,
+        path=path,
+        generated_at=generated_at,
+    )
+
+
+def _first_env(env: Mapping[str, str], *names: str) -> str:
+    for name in names:
+        value = env.get(name, "").strip()
+        if value:
+            return value
+    return ""
+
+
+def _local_artifact_digest(raw_path: str) -> str:
+    if not raw_path:
+        return ""
+    path = Path(raw_path).expanduser()
+    if not path.is_absolute():
+        path = (Path.cwd() / path).resolve()
+    if not path.exists():
+        return ""
+    digest = sha256()
+    if path.is_file():
+        _hash_file_into(digest, path, path.name)
+        return digest.hexdigest()
+    for file_path in sorted(item for item in path.rglob("*") if item.is_file()):
+        _hash_file_into(digest, file_path, file_path.relative_to(path).as_posix())
+    return digest.hexdigest()
+
+
+def _hash_file_into(digest: Any, file_path: Path, relative_name: str) -> None:
+    digest.update(relative_name.encode("utf-8"))
+    digest.update(b"\0")
+    with file_path.open("rb") as handle:
+        for chunk in iter(lambda: handle.read(1024 * 1024), b""):
+            digest.update(chunk)
+    digest.update(b"\0")
+
+
+def _canonical_json(payload: Any) -> str:
+    return json.dumps(payload, ensure_ascii=False, sort_keys=True, separators=(",", ":"))

static/app.js

@@ -300,15 +300,28 @@ function renderRefreshState(state) {
   }
   if (atlasRefreshProgressEl) {
     const show = status === "running" || status === "failed";
+    const cacheCopy = refreshQuestCacheCopy(state?.quest_cache || {});
     atlasRefreshProgressEl.hidden = !show;
     atlasRefreshProgressEl.textContent =
       status === "running"
-        ? `${stage || "Working"} · run ${state.run_id || ""}`
+        ? `${stage || "Working"}${cacheCopy ? ` · ${cacheCopy}` : ""} · run ${state.run_id || ""}`
         : state.error || "";
   }
   if (refreshDashboardButton) refreshDashboardButton.disabled = status === "running";
 }
 
+function refreshQuestCacheCopy(cache) {
+  const total = Number(cache.project_count || 0);
+  if (!total) return "";
+  const hits = Number(cache.hit_count || 0);
+  const misses = Number(cache.miss_count || 0);
+  const analyzed = Number(cache.analyzed_count || 0);
+  const remaining = Number(cache.remaining_count || 0);
+  if (!hits && !misses && !analyzed) return "";
+  if (remaining > 0) return `${hits} cached, ${analyzed}/${misses} analyzed`;
+  return `${hits} cached, ${analyzed} analyzed`;
+}
+
 function renderDashboard(data) {
   if (!data?.points?.length) {
     handleDashboardError(new Error("empty dashboard payload"));

tests/test_app.py

@@ -70,12 +70,15 @@ def _reset_refresh_state(status: str = "idle") -> None:
             {
                 "status": status,
                 "run_id": "test-run" if status == "running" else "",
+                "compute": "cpu" if status == "running" else "",
+                "reason": "test" if status == "running" else "",
                 "stage": "crawling" if status == "running" else "",
                 "stage_label": "Fetching public Spaces" if status == "running" else "",
                 "started_at": "",
                 "finished_at": "",
                 "error": "",
                 "result": None,
+                "quest_cache": app_module._empty_quest_cache_progress(),
             }
         )
 
@@ -179,6 +182,29 @@ def test_dashboard_refresh_rejects_concurrent_run(monkeypatch, tmp_path) -> None
         _reset_refresh_state()
 
 
+def test_dashboard_refresh_rejects_existing_bucket_lock(monkeypatch, tmp_path) -> None:
+    monkeypatch.setenv("ADVISOR_CACHE_DIR", str(tmp_path))
+    _reset_refresh_state()
+    (tmp_path / "refresh.lock").write_text(
+        json.dumps(
+            {
+                "run_id": "other-run",
+                "owner": "other-process",
+                "expires_at_epoch": time.time() + 3600,
+            }
+        ),
+        encoding="utf-8",
+    )
+
+    try:
+        dashboard_refresh_start()
+    except Exception as error:
+        assert getattr(error, "status_code", None) == 409
+        assert "other-run" in str(getattr(error, "detail", ""))
+    else:
+        raise AssertionError("dashboard refresh should honor an existing bucket lock")
+
+
 def test_dashboard_refresh_embedding_build_runs_in_subprocess(monkeypatch, tmp_path) -> None:
     project_path = tmp_path / "projects.json"
     index_path = tmp_path / "project_index.json"
@@ -237,7 +263,7 @@ def test_dashboard_refresh_persists_and_swaps_latest(monkeypatch, tmp_path) -> N
     monkeypatch.setenv("ADVISOR_CACHE_DIR", str(tmp_path))
     _reset_refresh_state()
 
-    def fake_refresh_payloads(run_id: str) -> tuple[dict, dict, dict]:
+    def fake_refresh_payloads(run_id: str, *, cache_dir, compute) -> tuple[dict, dict, dict, dict]:
         projects_payload = json.loads(app_module.DATA_PATH.read_text(encoding="utf-8"))
         index_payload = json.loads(app_module.INDEX_PATH.read_text(encoding="utf-8"))
         refreshed_index = ProjectIndex.from_files(app_module.DATA_PATH, app_module.INDEX_PATH)
@@ -245,7 +271,13 @@ def test_dashboard_refresh_persists_and_swaps_latest(monkeypatch, tmp_path) -> N
             refreshed_index,
             generated_at="2026-06-08T00:00:00+00:00",
         )
-        return projects_payload, index_payload, refreshed_dashboard
+        quest_analysis_payload = {
+            "schema_version": 1,
+            "run_id": run_id,
+            "summary": {"project_count": refreshed_dashboard["project_count"], "compute": compute},
+            "projects": [],
+        }
+        return projects_payload, index_payload, refreshed_dashboard, quest_analysis_payload
 
     monkeypatch.setattr(app_module, "_build_refresh_payloads", fake_refresh_payloads)
     response = dashboard_refresh_start()
@@ -254,11 +286,14 @@ def test_dashboard_refresh_persists_and_swaps_latest(monkeypatch, tmp_path) -> N
     state = _wait_for_refresh()
     assert state["status"] == "succeeded"
     assert (tmp_path / "latest.json").is_file()
+    assert (tmp_path / "refresh.lock").exists() is False
+    latest = json.loads((tmp_path / "latest.json").read_text(encoding="utf-8"))
+    assert (tmp_path / latest["quest_analysis"]).is_file()
     assert state["result"]["project_count"] == len(app_module.index.projects)
     assert dashboard()["provenance"]["snapshot_digest"] == state["result"]["snapshot_digest"]
 
 
-def test_dashboard_refresh_quest_analysis_uses_minicpm_analyzer(monkeypatch) -> None:
+def test_dashboard_refresh_quest_analysis_uses_minicpm_analyzer(monkeypatch, tmp_path) -> None:
     project = Project(
         id="build-small-hackathon/minicpm-refresh-smoke",
         title="MiniCPM Refresh Smoke",
@@ -301,14 +336,21 @@ def test_dashboard_refresh_quest_analysis_uses_minicpm_analyzer(monkeypatch) ->
 
     monkeypatch.setattr(app_module, "create_quest_analyzer", lambda device: FakeMiniCPMAnalyzer())
 
-    result = app_module._analyze_dashboard_quests([project.to_refresh_snapshot_dict()])
+    result = app_module._analyze_dashboard_quests(
+        [project.to_refresh_snapshot_dict()],
+        cache_dir=tmp_path,
+        compute="cpu",
+        run_id="test-run",
+    )
 
     quests = {match["quest"] for match in result["matches_by_project"][project.id]}
     assert result["source"] == "minicpm-json-quest-analyzer"
     assert quests == {"Off the Grid", "Field Notes"}
+    assert result["quest_analysis_payload"]["summary"]["miss_count"] == 1
+    assert result["quest_analysis_payload"]["summary"]["analyzed_count"] == 1
 
 
-def test_dashboard_refresh_quest_analysis_batches_minicpm(monkeypatch) -> None:
+def test_dashboard_refresh_quest_analysis_batches_minicpm(monkeypatch, tmp_path) -> None:
     projects = [
         Project(
             id=f"build-small-hackathon/batched-{index}",
@@ -341,7 +383,12 @@ def test_dashboard_refresh_quest_analysis_batches_minicpm(monkeypatch) -> None:
     monkeypatch.setenv("ADVISOR_QUEST_ANALYSIS_BATCH_SIZE", "2")
     monkeypatch.setattr(app_module, "create_quest_analyzer", lambda device: FakeMiniCPMAnalyzer())
 
-    result = app_module._analyze_dashboard_quests([project.to_refresh_snapshot_dict() for project in projects])
+    result = app_module._analyze_dashboard_quests(
+        [project.to_refresh_snapshot_dict() for project in projects],
+        cache_dir=tmp_path,
+        compute="cpu",
+        run_id="test-run",
+    )
 
     assert calls == [
         ["build-small-hackathon/batched-0", "build-small-hackathon/batched-1"],
@@ -350,7 +397,69 @@ def test_dashboard_refresh_quest_analysis_batches_minicpm(monkeypatch) -> None:
     assert set(result["matches_by_project"]) == {project.id for project in projects}
 
 
-def test_dashboard_refresh_quest_analysis_requires_two_segment_snapshot() -> None:
+def test_dashboard_refresh_quest_analysis_caches_minicpm_results(monkeypatch, tmp_path) -> None:
+    project = Project(
+        id="build-small-hackathon/cached-quest",
+        title="Cached Quest",
+        summary="A small local project.",
+        tags=("gradio",),
+        models=("openbmb/MiniCPM5-1B",),
+        datasets=(),
+        likes=0,
+        sdk="gradio",
+        license="mit",
+        created_at="2026-06-01T00:00:00+00:00",
+        last_modified="2026-06-08T00:00:00+00:00",
+        host="https://cached-quest.hf.space",
+        url="https://huggingface.co/spaces/build-small-hackathon/cached-quest",
+        readme_body="Runs MiniCPM5-1B locally.",
+        app_file_source="from transformers import AutoModelForCausalLM",
+    )
+    calls = []
+
+    class FakeMiniCPMAnalyzer:
+        source = "minicpm-json-quest-analyzer"
+
+        def analyze(self, projects):
+            calls.append([item.id for item in projects])
+            return {
+                project.id: [
+                    {
+                        "quest": "OpenBMB",
+                        "confidence": 0.91,
+                        "evidence": "Runs MiniCPM5-1B locally",
+                        "source": "readme",
+                    }
+                ]
+            }
+
+    monkeypatch.setattr(app_module, "create_quest_analyzer", lambda device: FakeMiniCPMAnalyzer())
+
+    first = app_module._analyze_dashboard_quests(
+        [project.to_refresh_snapshot_dict()],
+        cache_dir=tmp_path,
+        compute="cpu",
+        run_id="first-run",
+    )
+
+    def fail_analyzer(device):
+        raise AssertionError("cached quest analysis should not load MiniCPM")
+
+    monkeypatch.setattr(app_module, "create_quest_analyzer", fail_analyzer)
+    second = app_module._analyze_dashboard_quests(
+        [project.to_refresh_snapshot_dict()],
+        cache_dir=tmp_path,
+        compute="cpu",
+        run_id="second-run",
+    )
+
+    assert calls == [[project.id]]
+    assert first["matches_by_project"] == second["matches_by_project"]
+    assert second["quest_analysis_payload"]["summary"]["hit_count"] == 1
+    assert second["quest_analysis_payload"]["projects"][0]["status"] == "cached"
+
+
+def test_dashboard_refresh_quest_analysis_requires_two_segment_snapshot(tmp_path) -> None:
     project = Project(
         id="build-small-hackathon/missing-evidence",
         title="Missing Evidence",
@@ -372,7 +481,7 @@ def test_dashboard_refresh_quest_analysis_requires_two_segment_snapshot() -> Non
     del row["readme_body"]
 
     try:
-        app_module._analyze_dashboard_quests([row])
+        app_module._analyze_dashboard_quests([row], cache_dir=tmp_path, compute="cpu", run_id="test-run")
     except RuntimeError as error:
         assert "readme_body and app_file_source" in str(error)
     else:

tests/test_quest_cache.py

@@ -0,0 +1,96 @@
+from __future__ import annotations
+
+import json
+
+from hackathon_advisor.data import Project
+from hackathon_advisor.quest_cache import (
+    build_quest_cache_identity,
+    quest_analyzer_fingerprint_from_env,
+    read_quest_cache_entry,
+    write_quest_cache_entry,
+)
+
+
+def _project(readme_body: str = "Uses MiniCPM locally.") -> Project:
+    return Project(
+        id="build-small-hackathon/cache-unit",
+        title="Cache Unit",
+        summary="A small MiniCPM app.",
+        tags=("gradio",),
+        models=("openbmb/MiniCPM5-1B",),
+        datasets=(),
+        likes=0,
+        sdk="gradio",
+        license="mit",
+        created_at="2026-06-01T00:00:00+00:00",
+        last_modified="2026-06-08T00:00:00+00:00",
+        host="https://cache-unit.hf.space",
+        url="https://huggingface.co/spaces/build-small-hackathon/cache-unit",
+        app_file="app.py",
+        app_file_source="from transformers import AutoModelForCausalLM",
+        readme_body=readme_body,
+    )
+
+
+def test_quest_cache_key_changes_when_prompt_changes() -> None:
+    fingerprint = quest_analyzer_fingerprint_from_env({"ADVISOR_QUEST_ADAPTER_ID": ""})
+
+    first = build_quest_cache_identity(_project("Uses MiniCPM locally."), fingerprint)
+    second = build_quest_cache_identity(_project("Exports a PDF report."), fingerprint)
+
+    assert first.prompt_hash != second.prompt_hash
+    assert first.cache_key != second.cache_key
+
+
+def test_quest_cache_round_trip_validates_cached_matches(tmp_path) -> None:
+    project = _project()
+    fingerprint = quest_analyzer_fingerprint_from_env({"ADVISOR_QUEST_ADAPTER_ID": ""})
+    matches = [
+        {
+            "quest": "OpenBMB",
+            "confidence": 0.91,
+            "evidence": "Uses MiniCPM locally",
+            "source": "readme",
+        }
+    ]
+
+    stored = write_quest_cache_entry(
+        tmp_path,
+        project,
+        fingerprint,
+        matches,
+        source="minicpm-json-quest-analyzer",
+    )
+    lookup = read_quest_cache_entry(tmp_path, project, fingerprint)
+
+    assert lookup.reason == "hit"
+    assert lookup.entry is not None
+    assert lookup.entry.path == stored.path
+    assert lookup.entry.matches == stored.matches
+
+
+def test_quest_cache_rejects_corrupt_record(tmp_path) -> None:
+    project = _project()
+    fingerprint = quest_analyzer_fingerprint_from_env({"ADVISOR_QUEST_ADAPTER_ID": ""})
+    stored = write_quest_cache_entry(
+        tmp_path,
+        project,
+        fingerprint,
+        [
+            {
+                "quest": "OpenBMB",
+                "confidence": 0.91,
+                "evidence": "Uses MiniCPM locally",
+                "source": "readme",
+            }
+        ],
+        source="minicpm-json-quest-analyzer",
+    )
+    payload = json.loads(stored.path.read_text(encoding="utf-8"))
+    payload["matches"][0]["quest"] = "Unknown Quest"
+    stored.path.write_text(json.dumps(payload), encoding="utf-8")
+
+    lookup = read_quest_cache_entry(tmp_path, project, fingerprint)
+
+    assert lookup.entry is None
+    assert lookup.reason.startswith("invalid_schema:")