Deploy Transformers PR API

README.md

@@ -5,7 +5,7 @@ colorFrom: indigo
 colorTo: blue
 sdk: docker
 app_port: 7860
-short_description: Live API for Transformers PR similarity search.
+short_description: Live API for Transformers PR search and issue clustering.
 datasets:
   - evalstate/transformers-pr
 tags:
@@ -20,12 +20,6 @@ tags:
 
 Machine-oriented API for PR similarity search.
 
-Canonical storage roles:
-
-- dataset repo: published latest state and canonical current analysis
-- mounted bucket: mutable operational cache only
-- Space disk: ephemeral runtime storage
-
 Defaults for this deployment:
 
 - repo: `huggingface/transformers`

src/slop_farmer.egg-info/PKG-INFO

@@ -61,6 +61,16 @@ forward from the previous snapshot when the new snapshot does not already have i
 log a cache-hit summary for the run. This is useful for incremental scrapes where many
 review units are unchanged and can safely reuse cached hybrid decisions.
 
+To push that local cache back to the dataset repo for future remote-first runs, use either:
+
+- `publish-analysis-artifacts --save-cache` during canonical analysis publication
+- `save-cache` to upload `analysis-state/` on its own
+
+Hybrid review execution is bounded-parallel. Use `--hybrid-llm-concurrency N` or
+`analysis.hybrid_llm_concurrency: N` to cap concurrent review units. `1` keeps the
+lowest provider pressure; higher values can reduce wall-clock time at the cost of more
+provider pressure.
+
 ## Scope
 
 Cluster PRs by touched repository areas.
@@ -81,23 +91,40 @@ uv run slop-farmer scrape \
   --max-prs 50
 ```
 
-To publish a snapshot to the Hub:
+To refresh the canonical dataset repo:
 
 ```bash
-uv run slop-farmer scrape \
-  --repo huggingface/transformers \
-  --output-dir data \
-  --hf-repo-id burtenshaw/transformers-pr-slop-dataset \
-  --publish
+uv run slop-farmer --config configs/transformers.yaml refresh-dataset
 ```
 
-When `--publish` is used, `slop-farmer` now also generates and uploads new contributor reviewer artifacts by default:
+`refresh-dataset` publishes raw tables plus cheap artifacts like:
 
 - `new_contributors.parquet`
 - `new-contributors-report.json`
 - `new-contributors-report.md`
+- `pr-scope-clusters.json`
+
+To publish expensive hybrid analysis artifacts after a local `analyze` run:
+
+```bash
+uv run slop-farmer --config configs/transformers.yaml publish-analysis-artifacts \
+  --analysis-id hybrid-gpt54mini-v3 \
+  --canonical \
+  --save-cache
+```
+
+This writes an immutable archived run under
+`snapshots/<snapshot_id>/analysis-runs/<analysis_id>/...` and, with `--canonical`,
+updates the stable `analysis/current/` alias. With `--save-cache`, it also uploads the
+snapshot-local `analysis-state/` directory to repo-root `analysis-state/` as mutable
+operational cache for future hybrid runs.
+
+To upload only the cache without publishing canonical analysis:
 
-Use `--no-new-contributor-report` to skip them.
+```bash
+uv run slop-farmer --config configs/transformers.yaml save-cache \
+  --snapshot-dir runs/transformers-recent-60d/data/snapshots/20260418T170534Z
+```
 
 ## Nightly incremental runs
 
@@ -170,6 +197,7 @@ materialize versioned Hub **dataset repos**; it does not currently read HF bucke
 Compatibility wrappers remain available:
 
 - `scripts/submit_transformers_dataset_job.sh`
+- `scripts/submit_diffusers_dataset_job.sh`
 - `scripts/submit_openclaw_dataset_job.sh`
 
 For the current storage model and recommended modes, see
@@ -183,11 +211,15 @@ You can analyze the published Hugging Face dataset directly without scraping Git
 uv run slop-farmer analyze \                           
   --snapshot-dir eval_data/snapshots/gh-live-latest-1000x1000 \
   --ranking-backend hybrid \
-  --model "gpt-5-mini?reasoning=low" \
+  --model "gpt-5.4-mini?service_tier=flex" \
   --output /tmp/gh-live-latest-1000x1000-hybrid.json
 ```
 
-This materializes the dataset-viewer parquet export into a local snapshot cache under `eval_data/snapshots/` and writes `analysis-report.json` next to it.
+This materializes the dataset-viewer parquet export into a local snapshot cache under
+`eval_data/snapshots/` and writes a local analysis report next to it. Publishing
+canonical hybrid analysis is a separate `publish-analysis-artifacts` step, and updating
+the remote hybrid cache source is `publish-analysis-artifacts --save-cache` or
+standalone `save-cache`.
 
 Repo-local defaults for `analyze` can be stored in `pyproject.toml` under `[tool.slop-farmer.analyze]`. This repo currently defaults to:
 
@@ -283,28 +315,15 @@ By default this writes:
 
 next to the snapshot, including GitHub profile links, repo issue/PR search links, and example authored artifacts.
 
-## Full end-to-end workflow
-
-You can run scrape + publish + analyze + markdown + dashboard export in one command:
-
-```bash
-uv run slop-farmer full-pipeline \
-  --repo huggingface/transformers \
-  --dataset YOURNAME/transformers-pr-slop-dataset \
-  --model "gpt-5-mini?reasoning=low"
-```
+## Recommended end-to-end sequence
 
-This writes outputs under a repo-anchored workspace directory, for example:
+For canonical upkeep, prefer the explicit sequence:
 
-- `runs/transformers/data/`
-- `runs/transformers/web/public/data/`
-
-Optional age caps are based on `created_at`:
-
-```bash
-  --issue-max-age-days 30 \
-  --pr-max-age-days 14
-```
+1. `refresh-dataset`
+2. `analyze`
+3. `publish-analysis-artifacts --save-cache`
+4. `dashboard-data`
+5. deploy dashboard and API if needed
 
 ## Validation checks
 
@@ -312,6 +331,7 @@ Before committing or wiring new package moves into automation, run:
 
 ```bash
 uv run python scripts/enforce_packaging.py
+uv run python scripts/check_hf_cli_secrets.py
 uv run --extra dev ruff format --check src tests scripts jobs
 uv run --extra dev ruff check src tests scripts jobs
 uv run --extra dev ty check src tests scripts jobs
@@ -324,6 +344,9 @@ uv run --extra dev pytest -q
 - `data` must not import `reports`
 - `reports` must not import `app`
 
+`scripts/check_hf_cli_secrets.py` rejects `hf ... --secrets NAME=value` so access
+tokens cannot be exposed via process argv.
+
 ## YAML config-driven runs
 
 You can keep repo-specific pipeline defaults in a YAML file and apply them to all
@@ -378,9 +401,9 @@ uv run slop-farmer --config configs/diffusers.yaml dataset-status
 Those reader commands default to `dataset_id` when configured. Pass `--snapshot-dir` to force
 an explicit local snapshot instead.
 
-If you run `analyze` before `publish-snapshot`, the uploaded snapshot will also include
-`analysis-state/`, which makes the hybrid cache portable across machines and reusable in
-later snapshots when `analysis.cached_analysis: true` is enabled.
+`analysis-state/` is mutable operational cache only. You can upload it to the dataset
+repo with `save-cache` or `publish-analysis-artifacts --save-cache`, but it is still not
+the canonical analysis read surface.
 
 ## Export static dashboard data
 
@@ -402,6 +425,15 @@ This writes:
 
 The dashboard is intentionally summary-first and links out to GitHub for deep detail.
 
+When `--analysis-input` is omitted, `dashboard-data` now prefers:
+
+1. `analysis/current/manifest.json`
+2. `analysis/current/analysis-report-hybrid.json`
+3. snapshot-local fallback only when canonical current analysis is absent
+
+If the canonical current manifest exists but the required artifact is missing, dashboard export
+fails loudly instead of silently drifting to snapshot-local analysis.
+
 ## Deploy a dashboard to a Hugging Face Space
 
 Use the generic deploy script:
@@ -420,6 +452,12 @@ Repo-specific wrappers are also available:
 - `scripts/deploy_transformers_dashboard_space.sh`
 - `scripts/deploy_openclaw_dashboard_space.sh`
 
+Repo-specific end-to-end dashboard update helpers are also available:
+
+- `scripts/update_transformers_dashboard.sh`
+- `scripts/update_diffusers_dashboard.sh`
+- `scripts/update_openclaw_dashboard.sh`
+
 Or use the CLI wrapper with a YAML config:
 
 ```bash
@@ -432,21 +470,23 @@ The repo includes the FastAPI service for the read-oriented PR similarity surfac
 The standalone `pr-search` client now lives in the downstream `pr-search-cli`
 package.
 
-Deploy the OpenClaw API Space with:
+Repo-specific wrappers are available for the current deployed APIs:
 
 ```bash
+scripts/update_diffusers_pr_search_api.sh
+scripts/update_transformers_pr_search_api.sh
 scripts/update_openclaw_pr_search_api.sh
 ```
 
 Or use the generic deploy script directly:
 
 ```bash
-SPACE_ID=evalstate/openclaw-pr-api \
-SPACE_TITLE="OpenClaw PR API" \
-DEFAULT_REPO=openclaw/openclaw \
+SPACE_ID=evalstate/transformers-pr-api \
+SPACE_TITLE="Transformers PR API" \
+DEFAULT_REPO=huggingface/transformers \
 GHR_BASE_URL=https://ghreplica.dutiful.dev \
-HF_REPO_ID=evalstate/openclaw-pr \
-BUCKET_ID=evalstate/openclaw-pr-api-data \
+HF_REPO_ID=evalstate/transformers-pr \
+BUCKET_ID=evalstate/transformers-pr-api-data \
 scripts/deploy_pr_search_space.sh
 ```
 
@@ -455,14 +495,62 @@ This deploy flow:
 - creates or updates a Docker Space
 - uploads a minimal app bundle with a generated Space `README.md`
 - sets runtime variables for the API
-- mounts the configured HF bucket at `/data`
+- mounts the configured HF bucket at `/data` as mutable operational cache only
+
+Serving defaults:
+
+- dataset repo = canonical published state
+- API materializes one self-consistent dataset view
+- canonical `analysis/current/` is the default analysis surface when present
+- archived analysis is selectable explicitly with `snapshot_id` + `analysis_id`
 
 After the Space is live, you can query it either through the in-repo admin CLI:
 
 ```bash
-uv run slop-farmer pr-search status --repo openclaw/openclaw
-uv run slop-farmer pr-search similar 67096 --repo openclaw/openclaw
+uv run slop-farmer pr-search status --repo huggingface/transformers
+uv run slop-farmer pr-search similar 44940 --repo huggingface/transformers
 ```
 
 Or through the downstream `pr-search-cli` package, which owns the standalone
 `pr-search` executable.
+
+## Transformers migration cheat sheet
+
+To move Transformers onto the current architecture:
+
+### 1. Recreate the scheduled dataset refresh job with the generic wrapper
+
+```bash
+CONFIG_PATH=configs/transformers.yaml \
+LABEL=transformers-dataset-refresh \
+SCHEDULE='@daily' \
+scripts/submit_transformers_dataset_job.sh
+```
+
+This is the canonical scheduled writer for raw/latest dataset state.
+
+### 2. Run analysis and publish canonical hybrid analysis
+
+```bash
+ANALYSIS_ID=hybrid-gpt54mini-v3 scripts/update_transformers_dashboard.sh
+```
+
+That sequence:
+
+- refreshes dataset if requested
+- writes local hybrid analysis output
+- publishes canonical `analysis/current/`
+- saves repo-root `analysis-state/` for future hybrid cache reuse
+- rebuilds PR scope
+- deploys the dashboard
+
+### 3. Deploy the Transformers API Space
+
+```bash
+scripts/update_transformers_pr_search_api.sh
+```
+
+Optional runtime bucket:
+
+- default wrapper bucket id: `evalstate/transformers-pr-api-data`
+- treat it as mutable operational cache only, not canonical published storage

src/slop_farmer.egg-info/SOURCES.txt

@@ -19,9 +19,10 @@ src/slop_farmer/app/hf_checkpoint_import.py
 src/slop_farmer/app/pipeline.py
 src/slop_farmer/app/pr_search.py
 src/slop_farmer/app/pr_search_api.py
-src/slop_farmer/app/publish.py
+src/slop_farmer/app/publish_analysis.py
+src/slop_farmer/app/publish_dataset_snapshot.py
+src/slop_farmer/app/save_cache.py
 src/slop_farmer/app/snapshot_state.py
-src/slop_farmer/app/workflow.py
 src/slop_farmer/data/__init__.py
 src/slop_farmer/data/dataset_card.py
 src/slop_farmer/data/ghreplica_api.py
@@ -56,10 +57,12 @@ tests/test_cli.py
 tests/test_config.py
 tests/test_dashboard.py
 tests/test_dataset_status.py
+tests/test_deploy.py
 tests/test_farmer_setup_assets.py
 tests/test_ghreplica_api.py
 tests/test_github_api.py
 tests/test_hf_checkpoint_import.py
+tests/test_hf_cli_secrets_check.py
 tests/test_http.py
 tests/test_links.py
 tests/test_new_contributor_report.py
@@ -68,7 +71,11 @@ tests/test_pipeline_checkpoint_resume.py
 tests/test_pr_scope.py
 tests/test_pr_search.py
 tests/test_pr_search_api.py
-tests/test_publish.py
+tests/test_publish_analysis.py
+tests/test_published_layout.py
+tests/test_save_cache.py
 tests/test_snapshot_state.py
+tests/test_submit_dataset_job.py
+tests/test_update_dashboard_scripts.py
 tests/test_update_transformers_dataset.py
 tests/test_viewer_layout.py

src/slop_farmer/app/cli.py

@@ -23,6 +23,7 @@ from slop_farmer.config import (
     PrSearchRefreshOptions,
     PublishAnalysisArtifactsOptions,
     RepoRef,
+    SaveCacheOptions,
     SnapshotAdoptOptions,
 )
 from slop_farmer.reports.duplicate_prs import DEFAULT_DUPLICATE_PR_MODEL
@@ -63,6 +64,7 @@ def build_parser(*, config_path: Path | None = None) -> argparse.ArgumentParser:
     _add_new_contributor_report_parser(subparsers, defaults["new-contributor-report"])
     _add_dashboard_data_parser(subparsers, defaults["dashboard-data"])
     _add_publish_analysis_artifacts_parser(subparsers, defaults["publish-analysis-artifacts"])
+    _add_save_cache_parser(subparsers, defaults["save-cache"])
     _add_deploy_dashboard_parser(subparsers, defaults["deploy-dashboard"])
     _add_dataset_status_parser(subparsers, defaults["dataset-status"])
     return parser
@@ -80,6 +82,7 @@ def _load_parser_defaults(config_path: Path | None) -> dict[str, dict[str, Any]]
         "new-contributor-report",
         "dashboard-data",
         "publish-analysis-artifacts",
+        "save-cache",
         "deploy-dashboard",
         "dataset-status",
     )
@@ -897,6 +900,11 @@ def _add_publish_analysis_artifacts_parser(subparsers: Any, defaults: dict[str,
         type=Path,
         help="Optional explicit snapshot directory containing analysis-report-hybrid.json.",
     )
+    publish_analysis.add_argument(
+        "--analysis-input",
+        type=Path,
+        help="Optional explicit hybrid analysis report JSON to publish instead of snapshot-dir discovery.",
+    )
     publish_analysis.add_argument(
         "--hf-repo-id",
         default=defaults.get("hf-repo-id"),
@@ -910,6 +918,12 @@ def _add_publish_analysis_artifacts_parser(subparsers: Any, defaults: dict[str,
         default=bool(defaults.get("canonical", False)),
         help="Also update the stable analysis/current canonical alias.",
     )
+    publish_analysis.add_argument(
+        "--save-cache",
+        action="store_true",
+        default=bool(defaults.get("save-cache", False)),
+        help="Also upload snapshot-local analysis-state/ as mutable operational cache at repo-root analysis-state/.",
+    )
     publish_analysis.add_argument(
         "--private-hf-repo",
         action="store_true",
@@ -918,6 +932,36 @@ def _add_publish_analysis_artifacts_parser(subparsers: Any, defaults: dict[str,
     )
 
 
+def _add_save_cache_parser(subparsers: Any, defaults: dict[str, Any]) -> None:
+    save_cache = subparsers.add_parser(
+        "save-cache",
+        help="Upload snapshot-local analysis-state/ as mutable operational cache to a dataset repo.",
+    )
+    save_cache.add_argument(
+        "--output-dir",
+        type=Path,
+        default=Path(defaults.get("output-dir", "data")),
+        help="Pipeline workspace root containing snapshots/latest.json.",
+    )
+    save_cache.add_argument(
+        "--snapshot-dir",
+        type=Path,
+        help="Optional explicit snapshot directory containing analysis-state/.",
+    )
+    save_cache.add_argument(
+        "--hf-repo-id",
+        default=defaults.get("hf-repo-id"),
+        required=defaults.get("hf-repo-id") is None,
+        help="Target Hugging Face dataset repo id.",
+    )
+    save_cache.add_argument(
+        "--private-hf-repo",
+        action="store_true",
+        default=bool(defaults.get("private-hf-repo", False)),
+        help="Create the target dataset repo as private if needed.",
+    )
+
+
 def _add_deploy_dashboard_parser(subparsers: Any, defaults: dict[str, Any]) -> None:
     deploy_dashboard = subparsers.add_parser(
         "deploy-dashboard",
@@ -1512,9 +1556,30 @@ def _run_publish_analysis_artifacts(args: argparse.Namespace, config_path: Path
                 PublishAnalysisArtifactsOptions(
                     output_dir=args.output_dir,
                     snapshot_dir=args.snapshot_dir,
+                    analysis_input=args.analysis_input,
                     hf_repo_id=args.hf_repo_id,
                     analysis_id=args.analysis_id,
                     canonical=args.canonical,
+                    save_cache=args.save_cache,
+                    private_hf_repo=args.private_hf_repo,
+                )
+            ),
+            indent=2,
+        )
+    )
+
+
+def _run_save_cache(args: argparse.Namespace, config_path: Path | None) -> None:
+    del config_path
+    from slop_farmer.app.save_cache import run_save_cache
+
+    print(
+        json.dumps(
+            run_save_cache(
+                SaveCacheOptions(
+                    output_dir=args.output_dir,
+                    snapshot_dir=args.snapshot_dir,
+                    hf_repo_id=args.hf_repo_id,
                     private_hf_repo=args.private_hf_repo,
                 )
             ),
@@ -1543,6 +1608,7 @@ def main() -> None:
         "deploy-dashboard": _run_deploy_dashboard,
         "dataset-status": _run_dataset_status,
         "publish-analysis-artifacts": _run_publish_analysis_artifacts,
+        "save-cache": _run_save_cache,
     }
     handler = handlers.get(args.command)
     if handler is None:

src/slop_farmer/app/pr_search_api.py

@@ -1,7 +1,8 @@
 from __future__ import annotations
 
+import asyncio
 import os
-from contextlib import asynccontextmanager
+from contextlib import asynccontextmanager, suppress
 from dataclasses import dataclass
 from pathlib import Path
 from typing import Any, Literal
@@ -11,10 +12,19 @@ from fastapi.responses import JSONResponse
 
 from slop_farmer.config import PrSearchRefreshOptions
 from slop_farmer.data.ghreplica_api import GhReplicaProbeUnavailableError, GhrProbeClient
-from slop_farmer.data.snapshot_materialize import materialize_hf_dataset_snapshot
+from slop_farmer.data.snapshot_materialize import (
+    load_hf_current_analysis_manifest,
+    load_hf_current_search_manifest,
+    materialize_hf_current_analysis_surface,
+    materialize_hf_current_search_index,
+    materialize_hf_dataset_snapshot,
+)
 from slop_farmer.data.snapshot_paths import (
     CURRENT_ANALYSIS_MANIFEST_PATH,
     default_hf_materialize_dir,
+    load_current_analysis_manifest,
+    load_current_search_manifest,
+    repo_relative_path_to_local,
 )
 from slop_farmer.reports.analysis_service import (
     get_analysis_best,
@@ -64,6 +74,8 @@ class PrSearchApiSettings:
     http_max_retries: int = 5
     refresh_if_missing: bool = False
     rebuild_on_start: bool = False
+    current_analysis_poll_seconds: int = 0
+    current_search_poll_seconds: int = 0
     include_drafts: bool = False
     include_closed: bool = False
     similar_limit_default: int = 10
@@ -100,6 +112,8 @@ class PrSearchApiSettings:
             http_max_retries=_env_int("HTTP_MAX_RETRIES", 5),
             refresh_if_missing=_env_bool("REFRESH_IF_MISSING", False),
             rebuild_on_start=_env_bool("REBUILD_ON_START", False),
+            current_analysis_poll_seconds=_env_int("CURRENT_ANALYSIS_POLL_SECONDS", 0),
+            current_search_poll_seconds=_env_int("CURRENT_SEARCH_POLL_SECONDS", 0),
             include_drafts=_env_bool("INCLUDE_DRAFTS", False),
             include_closed=_env_bool("INCLUDE_CLOSED", False),
             similar_limit_default=_env_int("SIMILAR_LIMIT_DEFAULT", 10),
@@ -125,13 +139,28 @@ def create_app(settings: PrSearchApiSettings | None = None) -> FastAPI:
         app.state.settings = api_settings
         app.state.ready = False
         app.state.startup_error = None
+        app.state.current_analysis_refresh_error = None
+        app.state.current_search_refresh_error = None
+        refresh_task: asyncio.Task[None] | None = None
         try:
             _bootstrap_snapshot_assets(api_settings)
+            _bootstrap_current_search_index(api_settings)
             _bootstrap_index(api_settings)
             app.state.ready = _is_ready(api_settings)
+            if (
+                api_settings.current_analysis_poll_seconds > 0
+                or api_settings.current_search_poll_seconds > 0
+            ):
+                refresh_task = asyncio.create_task(_run_current_asset_refresh_loop(app))
         except Exception as exc:
             app.state.startup_error = str(exc)
-        yield
+        try:
+            yield
+        finally:
+            if refresh_task is not None:
+                refresh_task.cancel()
+                with suppress(asyncio.CancelledError):
+                    await refresh_task
 
     app = FastAPI(title="slop PR search API", version="0.1.1", lifespan=lifespan)
 
@@ -628,6 +657,84 @@ def _bootstrap_snapshot_assets(settings: PrSearchApiSettings) -> None:
     )
 
 
+def _bootstrap_current_search_index(settings: PrSearchApiSettings) -> None:
+    if settings.snapshot_dir is not None or settings.hf_repo_id is None:
+        return
+    _refresh_current_search_index(settings)
+
+
+async def _run_current_asset_refresh_loop(app: FastAPI) -> None:
+    settings = app.state.settings
+    interval = min_non_zero(
+        settings.current_analysis_poll_seconds,
+        settings.current_search_poll_seconds,
+    )
+    while True:
+        await asyncio.sleep(interval)
+        if settings.current_search_poll_seconds > 0:
+            try:
+                _refresh_current_search_index(settings)
+            except Exception as exc:
+                app.state.current_search_refresh_error = str(exc)
+            else:
+                app.state.current_search_refresh_error = None
+        if settings.current_analysis_poll_seconds > 0:
+            try:
+                _refresh_current_analysis_surface(settings)
+            except Exception as exc:
+                app.state.current_analysis_refresh_error = str(exc)
+            else:
+                app.state.current_analysis_refresh_error = None
+
+
+def _refresh_current_analysis_surface(settings: PrSearchApiSettings) -> bool:
+    if settings.hf_repo_id is None or settings.snapshot_dir is not None:
+        return False
+    remote_manifest = load_hf_current_analysis_manifest(
+        repo_id=settings.hf_repo_id,
+        revision=settings.hf_revision,
+    )
+    if remote_manifest is None:
+        return False
+    local_manifest = _load_local_current_analysis_manifest(settings)
+    if _analysis_manifest_identity(local_manifest) == _analysis_manifest_identity(remote_manifest):
+        return False
+    materialize_hf_current_analysis_surface(
+        repo_id=settings.hf_repo_id,
+        local_dir=_materialized_snapshot_dir(settings) or settings.output_dir,
+        revision=settings.hf_revision,
+    )
+    return True
+
+
+def _refresh_current_search_index(settings: PrSearchApiSettings) -> bool:
+    if settings.hf_repo_id is None or settings.snapshot_dir is not None:
+        return False
+    remote_manifest = load_hf_current_search_manifest(
+        repo_id=settings.hf_repo_id,
+        revision=settings.hf_revision,
+    )
+    if remote_manifest is None:
+        return False
+    local_manifest = _load_local_current_search_manifest(settings)
+    if _search_manifest_identity(local_manifest) == _search_manifest_identity(remote_manifest):
+        return False
+    manifest_path = _local_current_search_manifest_path(settings)
+    staged_db_path = settings.index_path.with_name(f".{settings.index_path.name}.download")
+    staged_manifest_path = manifest_path.with_name(f".{manifest_path.name}.download")
+    materialize_hf_current_search_index(
+        repo_id=settings.hf_repo_id,
+        db_path=staged_db_path,
+        manifest_path=staged_manifest_path,
+        revision=settings.hf_revision,
+    )
+    get_pr_search_status(staged_db_path, repo=settings.default_repo)
+    settings.index_path.parent.mkdir(parents=True, exist_ok=True)
+    staged_db_path.replace(settings.index_path)
+    staged_manifest_path.replace(manifest_path)
+    return True
+
+
 def _needs_refresh(settings: PrSearchApiSettings) -> bool:
     if settings.rebuild_on_start:
         return True
@@ -714,6 +821,64 @@ def _surface_available(snapshot_dir: Path, *, surface: Literal["issues", "contri
     return (snapshot_dir / "new-contributors-report.json").exists()
 
 
+def _load_local_current_analysis_manifest(settings: PrSearchApiSettings) -> dict[str, Any] | None:
+    materialized_snapshot_dir = _materialized_snapshot_dir(settings)
+    if materialized_snapshot_dir is None:
+        return None
+    manifest_path = repo_relative_path_to_local(
+        materialized_snapshot_dir, CURRENT_ANALYSIS_MANIFEST_PATH
+    )
+    if not manifest_path.exists():
+        return None
+    return load_current_analysis_manifest(manifest_path)
+
+
+def _local_current_search_manifest_path(settings: PrSearchApiSettings) -> Path:
+    return settings.index_path.with_name("current-search-manifest.json")
+
+
+def _load_local_current_search_manifest(settings: PrSearchApiSettings) -> dict[str, Any] | None:
+    manifest_path = _local_current_search_manifest_path(settings)
+    if not manifest_path.exists():
+        return None
+    return load_current_search_manifest(manifest_path)
+
+
+def _analysis_manifest_identity(
+    payload: dict[str, Any] | None,
+) -> tuple[str | None, str | None, str | None]:
+    if payload is None:
+        return (None, None, None)
+    snapshot_id = payload.get("snapshot_id")
+    analysis_id = payload.get("analysis_id")
+    published_at = payload.get("published_at")
+    return (
+        None if snapshot_id is None else str(snapshot_id),
+        None if analysis_id is None else str(analysis_id),
+        None if published_at is None else str(published_at),
+    )
+
+
+def _search_manifest_identity(
+    payload: dict[str, Any] | None,
+) -> tuple[str | None, str | None, str | None]:
+    if payload is None:
+        return (None, None, None)
+    snapshot_id = payload.get("snapshot_id")
+    run_id = payload.get("run_id")
+    published_at = payload.get("published_at")
+    return (
+        None if snapshot_id is None else str(snapshot_id),
+        None if run_id is None else str(run_id),
+        None if published_at is None else str(published_at),
+    )
+
+
+def min_non_zero(*values: int) -> int:
+    candidates = [value for value in values if value > 0]
+    return min(candidates) if candidates else 300
+
+
 def _limit(value: int | None, *, default: int, maximum: int) -> int:
     limit = default if value is None else value
     if limit < 1:

src/slop_farmer/app/publish_analysis.py

@@ -9,6 +9,7 @@ from typing import Any, Protocol, cast
 
 from huggingface_hub import CommitOperationAdd, HfApi, hf_hub_download
 
+from slop_farmer.app.save_cache import _save_analysis_cache_api
 from slop_farmer.config import PublishAnalysisArtifactsOptions
 from slop_farmer.data.parquet_io import read_json
 from slop_farmer.data.snapshot_paths import (
@@ -44,6 +45,16 @@ class HubApiLike(Protocol):
         repo_type: str,
     ) -> Any: ...
 
+    def upload_folder(
+        self,
+        *,
+        repo_id: str,
+        folder_path: Path,
+        path_in_repo: str,
+        repo_type: str,
+        commit_message: str,
+    ) -> None: ...
+
 
 @dataclass(frozen=True, slots=True)
 class PublishableAnalysisArtifacts:
@@ -59,9 +70,11 @@ def run_publish_analysis_artifacts(options: PublishAnalysisArtifactsOptions) ->
     snapshot_dir = resolve_snapshot_dir_from_output(options.output_dir, options.snapshot_dir)
     return publish_analysis_artifacts(
         snapshot_dir=snapshot_dir,
+        analysis_input=options.analysis_input,
         hf_repo_id=options.hf_repo_id,
         analysis_id=options.analysis_id,
         canonical=options.canonical,
+        save_cache=options.save_cache,
         private=options.private_hf_repo,
     )
 
@@ -69,19 +82,23 @@ def run_publish_analysis_artifacts(options: PublishAnalysisArtifactsOptions) ->
 def publish_analysis_artifacts(
     *,
     snapshot_dir: Path,
+    analysis_input: Path | None,
     hf_repo_id: str,
     analysis_id: str,
     canonical: bool,
     private: bool,
+    save_cache: bool = False,
     log: Callable[[str], None] | None = None,
 ) -> dict[str, Any]:
     return _publish_analysis_artifacts_api(
         cast("HubApiLike", HfApi()),
         snapshot_dir=snapshot_dir,
+        analysis_input=analysis_input,
         hf_repo_id=hf_repo_id,
         analysis_id=analysis_id,
         canonical=canonical,
         private=private,
+        save_cache=save_cache,
         log=log,
     )
 
@@ -90,13 +107,15 @@ def _publish_analysis_artifacts_api(
     api: HubApiLike,
     *,
     snapshot_dir: Path,
+    analysis_input: Path | None = None,
     hf_repo_id: str,
     analysis_id: str,
     canonical: bool,
     private: bool,
+    save_cache: bool = False,
     log: Callable[[str], None] | None = None,
 ) -> dict[str, Any]:
-    artifacts = _discover_publishable_analysis(snapshot_dir)
+    artifacts = _discover_publishable_analysis(snapshot_dir, analysis_input=analysis_input)
     published_at = _iso_now()
     channel = "canonical" if canonical else "comparison"
     archived_manifest = build_archived_analysis_run_manifest(
@@ -150,21 +169,37 @@ def _publish_analysis_artifacts_api(
         commit_message=f"Publish analysis {analysis_id} for snapshot {artifacts.snapshot_id}",
         repo_type="dataset",
     )
-    result = {
+    cache_result = (
+        _save_analysis_cache_api(
+            api,
+            snapshot_dir=snapshot_dir,
+            hf_repo_id=hf_repo_id,
+            private=private,
+            log=log,
+        )
+        if save_cache
+        else None
+    )
+    result: dict[str, Any] = {
         "repo": artifacts.repo,
         "dataset_id": hf_repo_id,
         "snapshot_id": artifacts.snapshot_id,
         "analysis_id": analysis_id,
         "canonical": canonical,
+        "save_cache": save_cache,
         "published_at": published_at,
         "artifact_paths": [operation.path_in_repo for operation in operations],
     }
+    if cache_result is not None:
+        result["cache"] = cache_result
     if log:
         log(f"Published analysis artifacts to {hf_repo_id}")
     return result
 
 
-def _discover_publishable_analysis(snapshot_dir: Path) -> PublishableAnalysisArtifacts:
+def _discover_publishable_analysis(
+    snapshot_dir: Path, *, analysis_input: Path | None
+) -> PublishableAnalysisArtifacts:
     manifest_path = snapshot_dir / ROOT_MANIFEST_FILENAME
     if not manifest_path.exists():
         raise FileNotFoundError(f"Snapshot manifest is missing: {manifest_path}")
@@ -176,7 +211,11 @@ def _discover_publishable_analysis(snapshot_dir: Path) -> PublishableAnalysisArt
     if not repo:
         raise ValueError(f"Snapshot manifest at {manifest_path} does not define repo.")
 
-    report_path = snapshot_dir / ANALYSIS_REPORT_FILENAME_BY_VARIANT["hybrid"]
+    report_path = (
+        analysis_input.resolve()
+        if analysis_input is not None
+        else snapshot_dir / ANALYSIS_REPORT_FILENAME_BY_VARIANT["hybrid"]
+    )
     if not report_path.exists():
         raise FileNotFoundError(f"Hybrid analysis report is missing: {report_path}")
     report_payload = read_json(report_path)
@@ -196,7 +235,7 @@ def _discover_publishable_analysis(snapshot_dir: Path) -> PublishableAnalysisArt
     if model is not None:
         model = str(model)
 
-    reviews_path = snapshot_dir / HYBRID_ANALYSIS_REVIEWS_FILENAME
+    reviews_path = report_path.with_name(f"{report_path.stem}.llm-reviews.json")
     return PublishableAnalysisArtifacts(
         repo=repo,
         snapshot_id=snapshot_id,
@@ -266,12 +305,13 @@ def _commit_operations(
     current_manifest: dict[str, Any] | None,
     snapshot_manifest: dict[str, Any],
 ) -> list[CommitOperationAdd]:
+    report_filename = ANALYSIS_REPORT_FILENAME_BY_VARIANT["hybrid"]
     operations = [
         CommitOperationAdd(
             path_in_repo=analysis_run_artifact_path(
                 artifacts.snapshot_id,
                 analysis_id,
-                artifacts.report_path.name,
+                report_filename,
             ),
             path_or_fileobj=artifacts.report_path,
         ),
@@ -290,7 +330,7 @@ def _commit_operations(
                 path_in_repo=analysis_run_artifact_path(
                     artifacts.snapshot_id,
                     analysis_id,
-                    artifacts.reviews_path.name,
+                    HYBRID_ANALYSIS_REVIEWS_FILENAME,
                 ),
                 path_or_fileobj=artifacts.reviews_path,
             )
@@ -299,7 +339,7 @@ def _commit_operations(
         operations.extend(
             [
                 CommitOperationAdd(
-                    path_in_repo=current_analysis_artifact_path(artifacts.report_path.name),
+                    path_in_repo=current_analysis_artifact_path(report_filename),
                     path_or_fileobj=artifacts.report_path,
                 ),
                 CommitOperationAdd(
@@ -311,7 +351,7 @@ def _commit_operations(
         if artifacts.reviews_path is not None:
             operations.append(
                 CommitOperationAdd(
-                    path_in_repo=current_analysis_artifact_path(artifacts.reviews_path.name),
+                    path_in_repo=current_analysis_artifact_path(HYBRID_ANALYSIS_REVIEWS_FILENAME),
                     path_or_fileobj=artifacts.reviews_path,
                 )
             )

src/slop_farmer/app/publish_pr_search_index.py

@@ -0,0 +1,141 @@
+from __future__ import annotations
+
+import json
+from collections.abc import Callable
+from dataclasses import dataclass
+from datetime import UTC, datetime
+from pathlib import Path
+from typing import Any, Protocol, cast
+
+from huggingface_hub import CommitOperationAdd, HfApi
+
+from slop_farmer.data.snapshot_paths import (
+    CURRENT_SEARCH_DB_PATH,
+    CURRENT_SEARCH_MANIFEST_PATH,
+    build_current_search_manifest,
+)
+from slop_farmer.reports.pr_search_service import get_pr_search_status
+
+
+class HubApiLike(Protocol):
+    def create_repo(
+        self,
+        repo_id: str,
+        *,
+        repo_type: str,
+        private: bool,
+        exist_ok: bool,
+    ) -> None: ...
+
+    def create_commit(
+        self,
+        repo_id: str,
+        operations: list[CommitOperationAdd],
+        *,
+        commit_message: str,
+        repo_type: str,
+    ) -> Any: ...
+
+
+@dataclass(frozen=True, slots=True)
+class PublishablePrSearchIndex:
+    repo: str
+    snapshot_id: str
+    run_id: str
+    db_path: Path
+    source_type: str | None
+    hf_repo_id: str | None
+    hf_revision: str | None
+    row_counts: dict[str, Any]
+
+
+def publish_current_pr_search_index(
+    *,
+    db_path: Path,
+    hf_repo_id: str,
+    private: bool = False,
+    log: Callable[[str], None] | None = None,
+) -> dict[str, Any]:
+    return _publish_current_pr_search_index_api(
+        cast("HubApiLike", HfApi()),
+        db_path=db_path,
+        hf_repo_id=hf_repo_id,
+        private=private,
+        log=log,
+    )
+
+
+def _publish_current_pr_search_index_api(
+    api: HubApiLike,
+    *,
+    db_path: Path,
+    hf_repo_id: str,
+    private: bool = False,
+    log: Callable[[str], None] | None = None,
+) -> dict[str, Any]:
+    publishable = _discover_publishable_pr_search_index(db_path)
+    published_at = _iso_now()
+    manifest = build_current_search_manifest(
+        repo=publishable.repo,
+        snapshot_id=publishable.snapshot_id,
+        run_id=publishable.run_id,
+        published_at=published_at,
+        source_type=publishable.source_type,
+        hf_repo_id=publishable.hf_repo_id,
+        hf_revision=publishable.hf_revision,
+        row_counts=publishable.row_counts,
+    )
+    operations = [
+        CommitOperationAdd(
+            path_in_repo=CURRENT_SEARCH_DB_PATH,
+            path_or_fileobj=publishable.db_path,
+        ),
+        CommitOperationAdd(
+            path_in_repo=CURRENT_SEARCH_MANIFEST_PATH,
+            path_or_fileobj=_json_bytes(manifest),
+        ),
+    ]
+    if log:
+        log(f"Ensuring Hub dataset repo exists: {hf_repo_id}")
+    api.create_repo(hf_repo_id, repo_type="dataset", private=private, exist_ok=True)
+    if log:
+        log(f"Publishing PR search index run {publishable.run_id} for {publishable.snapshot_id}")
+    api.create_commit(
+        hf_repo_id,
+        operations,
+        commit_message=f"Publish PR search index {publishable.run_id} for {publishable.snapshot_id}",
+        repo_type="dataset",
+    )
+    result = {
+        "repo": publishable.repo,
+        "dataset_id": hf_repo_id,
+        "snapshot_id": publishable.snapshot_id,
+        "run_id": publishable.run_id,
+        "published_at": published_at,
+        "artifact_paths": [operation.path_in_repo for operation in operations],
+    }
+    if log:
+        log(f"Published PR search index to {hf_repo_id}")
+    return result
+
+
+def _discover_publishable_pr_search_index(db_path: Path) -> PublishablePrSearchIndex:
+    status = get_pr_search_status(db_path)
+    return PublishablePrSearchIndex(
+        repo=str(status["repo"]),
+        snapshot_id=str(status["snapshot_id"]),
+        run_id=str(status["id"]),
+        db_path=db_path.resolve(),
+        source_type=(None if status.get("source_type") is None else str(status.get("source_type"))),
+        hf_repo_id=None if status.get("hf_repo_id") is None else str(status.get("hf_repo_id")),
+        hf_revision=(None if status.get("hf_revision") is None else str(status.get("hf_revision"))),
+        row_counts=dict(status.get("row_counts") or {}),
+    )
+
+
+def _json_bytes(payload: dict[str, Any]) -> bytes:
+    return (json.dumps(payload, indent=2, sort_keys=True) + "\n").encode("utf-8")
+
+
+def _iso_now() -> str:
+    return datetime.now(tz=UTC).replace(microsecond=0).isoformat().replace("+00:00", "Z")

src/slop_farmer/app/save_cache.py

@@ -0,0 +1,115 @@
+from __future__ import annotations
+
+from collections.abc import Callable
+from pathlib import Path
+from typing import Any, Protocol, cast
+
+from huggingface_hub import HfApi
+
+from slop_farmer.config import SaveCacheOptions
+from slop_farmer.data.parquet_io import read_json
+from slop_farmer.data.snapshot_paths import ROOT_MANIFEST_FILENAME, resolve_snapshot_dir_from_output
+
+ANALYSIS_STATE_DIRNAME = "analysis-state"
+
+
+class HubApiLike(Protocol):
+    def create_repo(
+        self,
+        repo_id: str,
+        *,
+        repo_type: str,
+        private: bool,
+        exist_ok: bool,
+    ) -> None: ...
+
+    def upload_folder(
+        self,
+        *,
+        repo_id: str,
+        folder_path: Path,
+        path_in_repo: str,
+        repo_type: str,
+        commit_message: str,
+    ) -> None: ...
+
+
+def run_save_cache(options: SaveCacheOptions) -> dict[str, Any]:
+    snapshot_dir = resolve_snapshot_dir_from_output(options.output_dir, options.snapshot_dir)
+    return save_analysis_cache(
+        snapshot_dir=snapshot_dir,
+        hf_repo_id=options.hf_repo_id,
+        private=options.private_hf_repo,
+    )
+
+
+def save_analysis_cache(
+    *,
+    snapshot_dir: Path,
+    hf_repo_id: str,
+    private: bool,
+    log: Callable[[str], None] | None = None,
+) -> dict[str, Any]:
+    return _save_analysis_cache_api(
+        cast("HubApiLike", HfApi()),
+        snapshot_dir=snapshot_dir,
+        hf_repo_id=hf_repo_id,
+        private=private,
+        log=log,
+    )
+
+
+def _save_analysis_cache_api(
+    api: HubApiLike,
+    *,
+    snapshot_dir: Path,
+    hf_repo_id: str,
+    private: bool,
+    log: Callable[[str], None] | None = None,
+) -> dict[str, Any]:
+    cache_dir = snapshot_dir / ANALYSIS_STATE_DIRNAME
+    if not cache_dir.exists():
+        raise FileNotFoundError(f"Analysis cache directory is missing: {cache_dir}")
+    if not cache_dir.is_dir():
+        raise NotADirectoryError(f"Analysis cache path is not a directory: {cache_dir}")
+    artifact_paths = _cache_artifact_paths(cache_dir)
+    if not artifact_paths:
+        raise ValueError(f"Analysis cache directory is empty: {cache_dir}")
+
+    manifest_path = snapshot_dir / ROOT_MANIFEST_FILENAME
+    manifest = read_json(manifest_path) if manifest_path.exists() else {}
+    if not isinstance(manifest, dict):
+        raise ValueError(f"Snapshot manifest at {manifest_path} must contain a JSON object.")
+    snapshot_id = str(manifest.get("snapshot_id") or snapshot_dir.name).strip()
+    repo = str(manifest.get("repo") or "").strip()
+
+    if log:
+        log(f"Ensuring Hub dataset repo exists: {hf_repo_id}")
+    api.create_repo(hf_repo_id, repo_type="dataset", private=private, exist_ok=True)
+    if log:
+        log(f"Saving analysis cache for snapshot {snapshot_id}")
+    api.upload_folder(
+        repo_id=hf_repo_id,
+        folder_path=cache_dir,
+        path_in_repo=ANALYSIS_STATE_DIRNAME,
+        repo_type="dataset",
+        commit_message=f"Save analysis cache for snapshot {snapshot_id}",
+    )
+    result = {
+        "dataset_id": hf_repo_id,
+        "snapshot_id": snapshot_id,
+        "artifact_paths": [f"{ANALYSIS_STATE_DIRNAME}/{path}" for path in artifact_paths],
+    }
+    if repo:
+        result["repo"] = repo
+    if log:
+        log(f"Saved analysis cache to {hf_repo_id}")
+    return result
+
+
+def _cache_artifact_paths(cache_dir: Path) -> list[str]:
+    return sorted(
+        str(path.relative_to(cache_dir).as_posix())
+        for path in cache_dir.rglob("*")
+        if path.is_file()
+    )

src/slop_farmer/app_config.py

@@ -234,6 +234,10 @@ def _dashboard_config_defaults(config_path: Path) -> dict[str, dict[str, Any]]:
             "output-dir": str(data_dir) if data_dir else None,
             "hf-repo-id": dataset_id,
         },
+        "save-cache": {
+            "output-dir": str(data_dir) if data_dir else None,
+            "hf-repo-id": dataset_id,
+        },
         "deploy-dashboard": {
             "pipeline-data-dir": str(data_dir) if data_dir else None,
             "web-dir": str(web_dir) if web_dir else None,

src/slop_farmer/config.py

@@ -244,9 +244,19 @@ class DatasetRefreshOptions:
 class PublishAnalysisArtifactsOptions:
     output_dir: Path
     snapshot_dir: Path | None
+    analysis_input: Path | None
     hf_repo_id: str
     analysis_id: str
     canonical: bool = False
+    save_cache: bool = False
+    private_hf_repo: bool = False
+
+
+@dataclass(slots=True)
+class SaveCacheOptions:
+    output_dir: Path
+    snapshot_dir: Path | None
+    hf_repo_id: str
     private_hf_repo: bool = False
 
 

src/slop_farmer/data/snapshot_materialize.py

@@ -15,6 +15,8 @@ from slop_farmer.data.parquet_io import read_json, write_text
 from slop_farmer.data.snapshot_paths import (
     CONTRIBUTOR_ARTIFACT_FILENAMES,
     CURRENT_ANALYSIS_MANIFEST_PATH,
+    CURRENT_SEARCH_DB_PATH,
+    CURRENT_SEARCH_MANIFEST_PATH,
     LEGACY_ANALYSIS_FILENAMES,
     PR_SCOPE_CLUSTERS_FILENAME,
     RAW_TABLE_FILENAMES,
@@ -24,6 +26,7 @@ from slop_farmer.data.snapshot_paths import (
     STATE_WATERMARK_PATH,
     load_archived_analysis_run_manifest,
     load_current_analysis_manifest,
+    load_current_search_manifest,
     repo_relative_path_to_local,
 )
 
@@ -64,6 +67,121 @@ def materialize_hf_dataset_snapshot(
     )
 
 
+def load_hf_current_analysis_manifest(
+    *,
+    repo_id: str,
+    revision: str | None = None,
+) -> dict[str, Any] | None:
+    try:
+        downloaded = Path(
+            hf_hub_download(
+                repo_id=repo_id,
+                repo_type="dataset",
+                filename=CURRENT_ANALYSIS_MANIFEST_PATH,
+                revision=revision,
+            )
+        )
+    except Exception:
+        return None
+    return load_current_analysis_manifest(downloaded)
+
+
+def materialize_hf_current_analysis_surface(
+    *,
+    repo_id: str,
+    local_dir: Path,
+    revision: str | None = None,
+) -> Path:
+    local_dir.mkdir(parents=True, exist_ok=True)
+    manifest = load_hf_current_analysis_manifest(repo_id=repo_id, revision=revision)
+    if manifest is None:
+        raise FileNotFoundError(
+            f"HF dataset {repo_id} does not expose {CURRENT_ANALYSIS_MANIFEST_PATH!r}"
+        )
+
+    staged_downloads: list[tuple[Path, Path]] = []
+
+    def stage(repo_path: str, *, required: bool) -> None:
+        try:
+            downloaded = Path(
+                hf_hub_download(
+                    repo_id=repo_id,
+                    repo_type="dataset",
+                    filename=repo_path,
+                    revision=revision,
+                )
+            )
+        except Exception:
+            if required:
+                raise
+            return
+        staged_downloads.append((downloaded, repo_relative_path_to_local(local_dir, repo_path)))
+
+    for repo_path in (
+        ROOT_MANIFEST_FILENAME,
+        SNAPSHOTS_LATEST_PATH,
+        "issues.parquet",
+        "pull_requests.parquet",
+    ):
+        stage(repo_path, required=False)
+
+    for repo_path in manifest.get("artifacts", {}).values():
+        if isinstance(repo_path, str) and repo_path:
+            stage(repo_path, required=True)
+
+    stage(CURRENT_ANALYSIS_MANIFEST_PATH, required=True)
+
+    for downloaded, destination in staged_downloads:
+        _copy_downloaded_file(downloaded, destination)
+
+    return local_dir
+
+
+def load_hf_current_search_manifest(
+    *,
+    repo_id: str,
+    revision: str | None = None,
+) -> dict[str, Any] | None:
+    try:
+        downloaded = Path(
+            hf_hub_download(
+                repo_id=repo_id,
+                repo_type="dataset",
+                filename=CURRENT_SEARCH_MANIFEST_PATH,
+                revision=revision,
+            )
+        )
+    except Exception:
+        return None
+    return load_current_search_manifest(downloaded)
+
+
+def materialize_hf_current_search_index(
+    *,
+    repo_id: str,
+    db_path: Path,
+    manifest_path: Path,
+    revision: str | None = None,
+) -> dict[str, Any]:
+    manifest = load_hf_current_search_manifest(repo_id=repo_id, revision=revision)
+    if manifest is None:
+        raise FileNotFoundError(
+            f"HF dataset {repo_id} does not expose {CURRENT_SEARCH_MANIFEST_PATH!r}"
+        )
+    db_repo_path = str(manifest.get("artifacts", {}).get("db") or CURRENT_SEARCH_DB_PATH)
+    downloaded_db = Path(
+        hf_hub_download(
+            repo_id=repo_id,
+            repo_type="dataset",
+            filename=db_repo_path,
+            revision=revision,
+        )
+    )
+    _copy_downloaded_file(downloaded_db, db_path)
+    write_text(json.dumps(manifest, indent=2, sort_keys=True) + "\n", manifest_path)
+    return manifest
+
+
 def _materialize_hf_snapshot_repo_snapshot(
     *,
     repo_id: str,

src/slop_farmer/data/snapshot_paths.py

@@ -48,6 +48,10 @@ LEGACY_ANALYSIS_FILENAMES: tuple[str, ...] = (
 CURRENT_ANALYSIS_DIR = PurePosixPath("analysis/current")
 CURRENT_ANALYSIS_MANIFEST_PATH = str(CURRENT_ANALYSIS_DIR / ROOT_MANIFEST_FILENAME)
 ANALYSIS_MANIFEST_SCHEMA_VERSION = 1
+CURRENT_SEARCH_DIR = PurePosixPath("search/current")
+CURRENT_SEARCH_MANIFEST_PATH = str(CURRENT_SEARCH_DIR / ROOT_MANIFEST_FILENAME)
+CURRENT_SEARCH_DB_PATH = str(CURRENT_SEARCH_DIR / "pr-search.duckdb")
+SEARCH_MANIFEST_SCHEMA_VERSION = 1
 
 
 @dataclass(frozen=True, slots=True)
@@ -90,6 +94,10 @@ def current_analysis_artifact_path(filename: str) -> str:
     return str(CURRENT_ANALYSIS_DIR / filename)
 
 
+def current_search_artifact_path(filename: str) -> str:
+    return str(CURRENT_SEARCH_DIR / filename)
+
+
 def repo_key(repo_slug: str) -> str:
     return _path_key(repo_slug)
 
@@ -195,6 +203,39 @@ def load_archived_analysis_run_manifest(path: Path) -> dict[str, Any]:
     return validate_archived_analysis_run_manifest(payload)
 
 
+def build_current_search_manifest(
+    *,
+    repo: str,
+    snapshot_id: str,
+    run_id: str,
+    published_at: str,
+    source_type: str | None,
+    hf_repo_id: str | None,
+    hf_revision: str | None,
+    row_counts: dict[str, Any] | None,
+) -> dict[str, Any]:
+    payload = {
+        "schema_version": SEARCH_MANIFEST_SCHEMA_VERSION,
+        "repo": repo,
+        "snapshot_id": snapshot_id,
+        "run_id": run_id,
+        "published_at": published_at,
+        "source_type": source_type,
+        "hf_repo_id": hf_repo_id,
+        "hf_revision": hf_revision,
+        "artifacts": {"db": CURRENT_SEARCH_DB_PATH},
+        "row_counts": row_counts or {},
+    }
+    return validate_current_search_manifest(payload)
+
+
+def load_current_search_manifest(path: Path) -> dict[str, Any]:
+    payload = read_json(path)
+    if not isinstance(payload, dict):
+        raise ValueError(f"Current search manifest at {path} must contain a JSON object.")
+    return validate_current_search_manifest(payload)
+
+
 def resolve_default_dashboard_analysis_report(
     snapshot_dir: Path,
 ) -> ResolvedAnalysisReportPath | None:
@@ -289,6 +330,52 @@ def validate_archived_analysis_run_manifest(payload: dict[str, Any]) -> dict[str
     return _validate_analysis_manifest(payload, require_archived_artifacts=False)
 
 
+def validate_current_search_manifest(payload: dict[str, Any]) -> dict[str, Any]:
+    schema_version = int(payload.get("schema_version", SEARCH_MANIFEST_SCHEMA_VERSION))
+    if schema_version != SEARCH_MANIFEST_SCHEMA_VERSION:
+        raise ValueError(
+            f"Current search manifest schema_version must be {SEARCH_MANIFEST_SCHEMA_VERSION}."
+        )
+    repo = str(payload.get("repo") or "").strip()
+    snapshot_id = str(payload.get("snapshot_id") or "").strip()
+    run_id = str(payload.get("run_id") or "").strip()
+    published_at = str(payload.get("published_at") or "").strip()
+    artifacts = payload.get("artifacts")
+    if not repo:
+        raise ValueError("Current search manifest must define repo.")
+    if not snapshot_id:
+        raise ValueError("Current search manifest must define snapshot_id.")
+    if not run_id:
+        raise ValueError("Current search manifest must define run_id.")
+    if not published_at:
+        raise ValueError("Current search manifest must define published_at.")
+    if not isinstance(artifacts, dict):
+        raise ValueError("Current search manifest must define an artifacts object.")
+    db_path = artifacts.get("db")
+    if db_path != CURRENT_SEARCH_DB_PATH:
+        raise ValueError(f"Current search manifest db artifact must be {CURRENT_SEARCH_DB_PATH!r}.")
+    return {
+        "schema_version": SEARCH_MANIFEST_SCHEMA_VERSION,
+        "repo": repo,
+        "snapshot_id": snapshot_id,
+        "run_id": run_id,
+        "published_at": published_at,
+        "source_type": (
+            None if payload.get("source_type") is None else str(payload.get("source_type"))
+        ),
+        "hf_repo_id": None if payload.get("hf_repo_id") is None else str(payload.get("hf_repo_id")),
+        "hf_revision": (
+            None if payload.get("hf_revision") is None else str(payload.get("hf_revision"))
+        ),
+        "artifacts": {"db": CURRENT_SEARCH_DB_PATH},
+        "row_counts": (
+            {str(key): value for key, value in payload.get("row_counts", {}).items()}
+            if isinstance(payload.get("row_counts"), dict)
+            else {}
+        ),
+    }
+
+
 def load_latest_snapshot_pointer(snapshots_root: Path) -> Path | None:
     resolved_snapshots_root = snapshots_root.resolve()
     latest_path = resolved_snapshots_root / "latest.json"