Deploy simready-assets run 20261005-ee37df6 (GPU variant) (part 2)

pipeline/bake.py

@@ -19,6 +19,7 @@ import subprocess
 from pathlib import Path
 from typing import Dict, List, Optional
 
+from . import hf_addressing
 from .artifact_store import artifact_env
 from .config import (
     BUILD_DIR,
@@ -185,6 +186,32 @@ def _image_names(cfg: Config) -> Dict[str, str]:
     return names
 
 
+def url_mapping_rules(hf_repos: List[List[str]]) -> List[Dict]:
+    """deepsearch-api's USDSEARCH_URL_MAPPING_RULES for the HF repos a run
+    indexed (manifest["hf_repos"]: host, repo_type, repo_id, revision): each
+    repo's hf:// prefix -> its browsable huggingface.co file-viewer prefix.
+
+    A pinned revision maps `hf://.../resolve/<sha>/` to `.../blob/<sha>/`, so
+    the links show exactly what was indexed. `.../blob/main/`, the public form
+    an unpinned corpus emits, stays accepted inbound as an alias (never
+    emitted), so links handed out before the pin still resolve. Unpinned
+    (`main`) keeps the plain `hf://.../<repo>/` -> `.../blob/main/` rule.
+    """
+    rules = []
+    for host, repo_type, repo_id, revision in hf_repos:
+        prefix = hf_addressing.REPO_TYPE_PREFIX.get(repo_type, "")
+        web = f"https://{host}/" + "/".join(p for p in (prefix, repo_id) if p)
+        rule = {
+            "internal": hf_addressing.build(host=host, repo_type=repo_type, repo_id=repo_id,
+                                            revision=revision, is_folder=True),
+            "public": f"{web}/blob/{revision}/",
+        }
+        if hf_addressing.looks_like_sha(revision):
+            rule["aliases"] = [f"{web}/blob/main/"]
+        rules.append(rule)
+    return rules
+
+
 def hf_space_build_args(cfg: Config, manifest: Dict) -> Dict[str, str]:
     """The `docker/Dockerfile.hf-space` ARGs this run needs, shared between
     the local/CI `docker build` (bake.run, as --build-arg) and pipeline/
@@ -216,6 +243,13 @@ def hf_space_build_args(cfg: Config, manifest: Dict) -> Dict[str, str]:
         "SIGLIP2_NVCF_ENDPOINT": cfg.get("SIGLIP2_NVCF_ENDPOINT", SIGLIP2_NVCF_ENDPOINT_DEFAULT),
         "SIGLIP2_NVCF_FUNCTION_ID": cfg.get("SIGLIP2_NVCF_FUNCTION_ID", SIGLIP2_NVCF_FUNCTION_ID_DEFAULT),
     }
+    # The hf:// -> https:// link rewrite (url_mapping_rules) comes from the
+    # repos this run actually indexed, at their pinned revision, so it can't
+    # go stale when the corpus is re-pinned. As a Space Variable it is both
+    # the build-arg and the runtime env var. Off for an s3:// corpus.
+    rules = url_mapping_rules(manifest.get("hf_repos") or []) if storage_backend == "storage_api" else []
+    args["USDSEARCH_URL_MAPPING_MODE"] = "rules" if rules else "off"
+    args["USDSEARCH_URL_MAPPING_RULES"] = json.dumps(rules, separators=(",", ":"))
     # S3_STORAGE_* is only ever meaningful for an S3-sourced corpus, and
     # threading a source bucket name through for an HF-sourced one would be
     # actively misleading.
@@ -395,6 +429,48 @@ def download_smoke(base_url: str, timeout_s: float = 180.0) -> bool:
     return False
 
 
+def public_url_smoke(base_url: str, public_prefixes: List[str], timeout_s: float = 180.0) -> tuple:
+    """Every asset URL the Space hands out is a public huggingface.co link
+    under this run's URL-mapping rule: v2 search `url`s, v3 hybrid
+    `base_key`s and a download manifest's file URLs. A bare `https://` check
+    is not enough. The stale hand-set rule emitted
+    `https://.../blob/main/resolve/<sha>/...` (a 404), which passes it, so
+    each URL must start with one of `public_prefixes`
+    (url_mapping_rules()'s `public`s). Returns (ok, detail). Polls like
+    download_smoke(): the seed restores the index after /health is up."""
+    import time
+    import httpx
+
+    def bad(urls):
+        return [u for u in urls if not (isinstance(u, str) and u.startswith("https://")
+                                        and u.startswith(tuple(public_prefixes)))]
+
+    detail = "no URLs returned"
+    deadline = time.monotonic() + timeout_s
+    while time.monotonic() < deadline:
+        try:
+            v2 = httpx.get(f"{base_url}/search", params={"description": "", "limit": 25, "return_images": "false"},
+                           timeout=30.0).json()
+            v2 = v2 if isinstance(v2, list) else v2.get("hits", [])
+            v3 = httpx.post(f"{base_url}/search_hybrid", json={"description": "", "limit": 25, "return_images": False},
+                            timeout=30.0).json().get("hits", [])
+            urls = [h.get("url") for h in v2] + [(h.get("source") or {}).get("base_key") for h in v3]
+            if urls and urls[0]:
+                files = httpx.get(f"{base_url}/download/asset", params={"asset_url": urls[0], "manifest_only": "true"},
+                                  timeout=60.0).json().get("files", [])
+                urls += [f.get("url") for f in files]
+                wrong = bad(urls)
+                if not wrong and v2 and v3 and files:
+                    return True, f"{len(urls)} URLs, all under {', '.join(public_prefixes)}"
+                if wrong:
+                    return False, f"{len(wrong)}/{len(urls)} URLs not under {', '.join(public_prefixes)}, e.g. {wrong[0]}"
+                detail = f"incomplete response (v2 {len(v2)}, v3 {len(v3)}, download files {len(files)})"
+        except (httpx.HTTPError, ValueError, KeyError, IndexError, TypeError, AttributeError) as exc:
+            detail = f"{type(exc).__name__}: {exc}"
+        time.sleep(3)
+    return False, detail
+
+
 def verify_baked(cfg: Config, tags: Dict[str, str], manifest: Dict) -> int:
     """Check the data from a fresh `docker run`, not from the build log.
 
@@ -489,7 +565,8 @@ def verify_baked(cfg: Config, tags: Dict[str, str], manifest: Dict) -> int:
         # this smoke test is what actually proves that path works, not just
         # that the image builds. A longer deadline than the other images'
         # health checks: startup now includes a real S3 download + bulk
-        # load, not just a process coming up.
+        # load, not just a process coming up. RESTORE_RUN_ID pins the restore
+        # to this run: latest.json may name another one (`upload --no-latest`).
         token = cfg.siglip2_nvcf_env()["SIGLIP2_TRITON_SERVER_AUTH_TOKEN"]
         aws = artifact_env()
         result = subprocess.run(
@@ -497,6 +574,7 @@ def verify_baked(cfg: Config, tags: Dict[str, str], manifest: Dict) -> int:
              "-e", f"SIGLIP2_TRITON_SERVER_AUTH_TOKEN={token}",
              "-e", f"SIMREADYCENTRAL_INDEX_STORE_AWS_ACCESS_KEY_ID={aws['AWS_ACCESS_KEY_ID']}",
              "-e", f"SIMREADYCENTRAL_INDEX_STORE_AWS_SECRET_ACCESS_KEY={aws['AWS_SECRET_ACCESS_KEY']}",
+             "-e", f"RESTORE_RUN_ID={manifest['run_id']}",
              tags["hf_space"]],
             capture_output=True, text=True, check=True, timeout=60,
         )
@@ -569,7 +647,18 @@ def verify_baked(cfg: Config, tags: Dict[str, str], manifest: Dict) -> int:
                     print("    FAIL: hf-space image's download route (hf-storage-api) does not answer")
                     ok = False
 
-            manifest["hf_space_smoke_passed"] = healthy and graph_ok and download_ok
+            # Returned asset URLs are this run's public huggingface.co links
+            # (USDSEARCH_URL_MAPPING_RULES), so a stale or missing rule can't ship.
+            urls_ok = True
+            rules = url_mapping_rules(manifest.get("hf_repos") or [])
+            if healthy and manifest.get("storage_backend") == "storage_api" and rules:
+                urls_ok, detail = public_url_smoke("http://localhost:17860", [r["public"] for r in rules])
+                print(f"    returned asset URLs are public https:// links: {urls_ok} ({detail})")
+                if not urls_ok:
+                    print("    FAIL: hf-space image returns asset URLs outside this run's public URL mapping")
+                    ok = False
+
+            manifest["hf_space_smoke_passed"] = healthy and graph_ok and download_ok and urls_ok
             save_manifest(manifest)
         finally:
             subprocess.run(["docker", "rm", "-f", container], capture_output=True, check=False, timeout=30)
@@ -590,6 +679,7 @@ def verify_baked(cfg: Config, tags: Dict[str, str], manifest: Dict) -> int:
             ["docker", "run", "--rm", "-d", "-p", "17861:7860", "--gpus", "all",
              "-e", f"SIMREADYCENTRAL_INDEX_STORE_AWS_ACCESS_KEY_ID={aws['AWS_ACCESS_KEY_ID']}",
              "-e", f"SIMREADYCENTRAL_INDEX_STORE_AWS_SECRET_ACCESS_KEY={aws['AWS_SECRET_ACCESS_KEY']}",
+             "-e", f"RESTORE_RUN_ID={manifest['run_id']}",
              tags["hf_space_gpu"]],
             capture_output=True, text=True, check=True, timeout=60,
         )
@@ -647,7 +737,18 @@ def verify_baked(cfg: Config, tags: Dict[str, str], manifest: Dict) -> int:
                     print("    FAIL: hf-space-gpu image's download route (hf-storage-api) does not answer")
                     ok = False
 
-            manifest["hf_space_gpu_smoke_passed"] = healthy and embedding_ok and download_ok
+            # Returned asset URLs are this run's public huggingface.co links
+            # (USDSEARCH_URL_MAPPING_RULES), so a stale or missing rule can't ship.
+            urls_ok = True
+            rules = url_mapping_rules(manifest.get("hf_repos") or [])
+            if healthy and manifest.get("storage_backend") == "storage_api" and rules:
+                urls_ok, detail = public_url_smoke("http://localhost:17861", [r["public"] for r in rules])
+                print(f"    returned asset URLs are public https:// links: {urls_ok} ({detail})")
+                if not urls_ok:
+                    print("    FAIL: hf-space-gpu image returns asset URLs outside this run's public URL mapping")
+                    ok = False
+
+            manifest["hf_space_gpu_smoke_passed"] = healthy and embedding_ok and download_ok and urls_ok
             save_manifest(manifest)
         finally:
             subprocess.run(["docker", "rm", "-f", container], capture_output=True, check=False, timeout=30)

pipeline/restore.py

@@ -8,7 +8,10 @@ cleared first — see `clear.py`), or `removed` (cleared and never resubmitted).
 
 Skipped cleanly on the very first run for a corpus: no `latest.json` yet and
 no explicit `RESTORE_RUN_ID` means there is nothing to restore, so the stack
-just starts empty, exactly as before this feature existed.
+just starts empty, exactly as before this feature existed. `--no-restore`
+forces that same empty start for a corpus that does have previous runs — a
+full rebuild, e.g. after every asset URL changed (a new pinned revision), where
+a restored baseline would keep each asset's old URL alongside its new one.
 
 The OpenSearch load mirrors `docker/opensearch-seed.sh` (same PUT-index +
 `_bulk` + `_refresh` sequence, same settings/mappings merge from `bake.py`),
@@ -201,10 +204,13 @@ def restore_neo4j(cfg: Config, manifest: Dict) -> None:
           "start", "neo4j", "asset-graph-service"], env=cfg.compose_env())
 
 
-def run(cfg: Config) -> int:
-    prefix = resolve_baseline_prefix(cfg)
+def run(cfg: Config, fresh: bool = False) -> int:
+    prefix = None if fresh else resolve_baseline_prefix(cfg)
     if prefix is None:
-        print("no previous run found for this corpus — starting from an empty index")
+        if fresh:
+            print("--no-restore — starting from an empty index, ignoring any previous run")
+        else:
+            print("no previous run found for this corpus — starting from an empty index")
         if RESTORE_DIR.exists():
             shutil.rmtree(RESTORE_DIR)
         RESTORE_DIR.mkdir(parents=True)

pipeline/run_pipeline.py

@@ -115,7 +115,7 @@ def stage_down(cfg: Config, args) -> int:
 
 def stage_restore(cfg: Config, args) -> int:
     from .restore import run
-    return run(cfg)
+    return run(cfg, fresh=getattr(args, "no_restore", False))
 
 
 def stage_clear(cfg: Config, args) -> int:
@@ -158,7 +158,7 @@ def stage_export(cfg: Config, args) -> int:
 
 def stage_upload(cfg: Config, args) -> int:
     from .upload import run
-    return run(cfg)
+    return run(cfg, update_latest=not getattr(args, "no_latest", False))
 
 
 def stage_bake(cfg: Config, args) -> int:
@@ -343,6 +343,10 @@ def main(argv: List[str] | None = None) -> int:
                         help="hierarchy: place only Q-codes the tree has never seen; nothing placed moves (what `all` runs)")
     parser.add_argument("--no-upload", action="store_true",
                         help="hierarchy: write .build/hierarchy/ only, publish nothing to the artifact bucket")
+    parser.add_argument("--no-restore", action="store_true",
+                        help="restore/index/all: start from an empty index instead of the previous run (full rebuild)")
+    parser.add_argument("--no-latest", action="store_true",
+                        help="upload: publish the run's artifacts but leave latest.json (what Spaces load) unchanged")
     parser.add_argument("--start", choices=list(STAGES), default=None,
                         help="all: resume from this stage")
     parser.add_argument("--stop", choices=list(STAGES), default=None,

pipeline/submission_corpus.py

@@ -0,0 +1,224 @@
+"""Build a `storage_api` corpus file from a SimReady Central submission list.
+
+A submission list (e.g. `submissions/nvidia/simready-assets` in
+github.com/NVIDIA-Omniverse/simready-central) is a plain-text file with one
+HF URL per line, each naming one package's marker at a pinned commit:
+
+    https://huggingface.co/datasets/nvidia/simready-assets/resolve/<sha>/<pkg>/com.nvidia.simready.packaging.json
+
+It is the curated, pre-validated set of packages that should be in the index —
+the HF dataset's `main` keeps growing past it. This module turns that list into
+the corpus's `include` (one entry USD per package, pinned to the list's sha so
+the index never silently tracks `main`), keeping every other corpus key from
+`--base`:
+
+    python -m pipeline.submission_corpus --base config/corpus.simready-assets.json
+    python -m pipeline.submission_corpus --base config/corpus.simready-assets.json \\
+        --from-file /tmp/simready-assets --verify
+
+Reading from GitHub shells out to `gh api` (the simready-central repo is
+private), so `gh auth login` must have been done; `--from-file` needs nothing.
+The GitHub commit the list was read from is recorded under the corpus's
+`source` key (ignored by `Corpus.load`) so a corpus can be traced back to it.
+"""
+
+from __future__ import annotations
+
+import argparse
+import base64
+import json
+import re
+import subprocess
+import sys
+from dataclasses import dataclass
+from pathlib import Path
+from typing import Dict, List, Optional
+
+from pipeline import hf_addressing
+from pipeline.corpus import DEFAULT_ASSET_MARKER, Corpus
+
+DEFAULT_GITHUB_REPO = "NVIDIA-Omniverse/simready-central"
+DEFAULT_GITHUB_PATH = "submissions/nvidia/simready-assets"
+DEFAULT_GITHUB_REF = "main"
+
+# The root USD of a SimReady package — the same naming convention the
+# simready-assets corpus's `content_fallback` already encodes.
+DEFAULT_ENTRY_TEMPLATE = "{pkg}/simready_usd/sm_{pkg}_01.usd"
+
+_LINE_RE = re.compile(
+    r"^https://(?P<host>[^/]+)/datasets/(?P<repo_id>[^/]+/[^/]+)"
+    r"/resolve/(?P<revision>[^/]+)/(?P<pkg>[^/]+)/(?P<marker>[^/]+)$"
+)
+
+
+class SubmissionError(ValueError):
+    """The submission list cannot be turned into a corpus as written."""
+
+
+@dataclass(frozen=True)
+class Submission:
+    host: str
+    repo_id: str
+    revision: str
+    packages: List[str]  # in list order
+
+
+def parse_submission(text: str, marker: str = DEFAULT_ASSET_MARKER) -> Submission:
+    """Parse and validate a submission list. Blank lines and `#` comments are
+    skipped; every other line must name `marker` directly under a top-level
+    package folder, all in one dataset repo at one pinned commit sha."""
+    repos = set()
+    packages: List[str] = []
+    seen: Dict[str, int] = {}
+    for lineno, raw in enumerate(text.splitlines(), 1):
+        line = raw.strip()
+        if not line or line.startswith("#"):
+            continue
+        m = _LINE_RE.match(line)
+        if not m:
+            raise SubmissionError(
+                f"line {lineno}: {line!r} is not "
+                f"https://<host>/datasets/<owner>/<name>/resolve/<sha>/<package>/<marker>"
+            )
+        if m["marker"] != marker:
+            raise SubmissionError(f"line {lineno}: names {m['marker']!r}, expected the package marker {marker!r}")
+        if not hf_addressing.looks_like_sha(m["revision"]):
+            raise SubmissionError(
+                f"line {lineno}: /resolve/{m['revision']}/ is not a full commit sha — "
+                f"a submission list must pin one"
+            )
+        repos.add((m["host"], m["repo_id"], m["revision"].lower()))
+        if m["pkg"] in seen:
+            raise SubmissionError(f"line {lineno}: package {m['pkg']!r} already listed on line {seen[m['pkg']]}")
+        seen[m["pkg"]] = lineno
+        packages.append(m["pkg"])
+    if not packages:
+        raise SubmissionError("submission list names no packages")
+    if len(repos) != 1:
+        raise SubmissionError(
+            "submission list spans more than one repo/revision: "
+            + ", ".join(f"{host}/{repo}@{rev}" for host, repo, rev in sorted(repos))
+        )
+    (host, repo_id, revision), = repos
+    return Submission(host=host, repo_id=repo_id, revision=revision, packages=packages)
+
+
+def entry_paths(sub: Submission, entry_template: str = DEFAULT_ENTRY_TEMPLATE) -> List[str]:
+    """Repo-relative path of each package's entry USD, sorted."""
+    return sorted(entry_template.format(pkg=pkg) for pkg in sub.packages)
+
+
+def build_corpus(
+    base: dict,
+    sub: Submission,
+    entry_template: str = DEFAULT_ENTRY_TEMPLATE,
+    source: Optional[dict] = None,
+) -> dict:
+    """`base` with `include` replaced by the submission's pinned entry USDs.
+    Validated through `Corpus.from_dict`, so a result that `enumerate` would
+    reject never gets written."""
+    if base.get("storage_backend") != "storage_api":
+        raise SubmissionError("base corpus must have storage_backend: storage_api")
+    doc = dict(base)
+    doc["include"] = [
+        hf_addressing.build(
+            host=sub.host, repo_type="dataset", repo_id=sub.repo_id,
+            revision=sub.revision, path=path,
+        )
+        for path in entry_paths(sub, entry_template)
+    ]
+    if source is not None:
+        doc["source"] = source
+    else:
+        doc.pop("source", None)
+    Corpus.from_dict(doc)
+    return doc
+
+
+def verify_against_repo(sub: Submission, entry_template: str = DEFAULT_ENTRY_TEMPLATE,
+                        marker: str = DEFAULT_ASSET_MARKER) -> List[str]:
+    """Every listed package's marker and entry USD missing from the repo at
+    the pinned sha (empty list = all present). One recursive tree listing."""
+    from huggingface_hub import HfApi, RepoFile
+
+    present = {
+        entry.path
+        for entry in HfApi(endpoint=f"https://{sub.host}").list_repo_tree(
+            sub.repo_id, recursive=True, revision=sub.revision, repo_type="dataset",
+        )
+        if isinstance(entry, RepoFile)
+    }
+    missing = []
+    for pkg in sub.packages:
+        for path in (f"{pkg}/{marker}", entry_template.format(pkg=pkg)):
+            if path not in present:
+                missing.append(path)
+    return missing
+
+
+def _gh_api(endpoint: str, jq: str) -> str:
+    try:
+        out = subprocess.run(
+            ["gh", "api", endpoint, "--jq", jq], check=True, capture_output=True, text=True,
+        )
+    except FileNotFoundError:
+        raise SystemExit("`gh` not found — install the GitHub CLI, or pass --from-file")
+    except subprocess.CalledProcessError as exc:
+        raise SystemExit(f"gh api {endpoint} failed: {exc.stderr.strip()}")
+    return out.stdout.strip()
+
+
+def fetch_from_github(repo: str, path: str, ref: str) -> tuple[str, dict]:
+    """The list's text and a `source` record pinning the exact commit it was
+    read from (so `--ref main` is still reproducible after the fact)."""
+    commit = _gh_api(f"repos/{repo}/commits/{ref}", ".sha")
+    content = _gh_api(f"repos/{repo}/contents/{path}?ref={commit}", ".content")
+    text = base64.b64decode(content).decode("utf-8")
+    return text, {"github": f"https://github.com/{repo}/blob/{commit}/{path}", "commit": commit}
+
+
+def main(argv: Optional[List[str]] = None) -> int:
+    ap = argparse.ArgumentParser(description=__doc__.split("\n\n")[0])
+    ap.add_argument("--base", type=Path, required=True,
+                    help="existing corpus JSON whose non-include keys are kept")
+    ap.add_argument("--out", type=Path, help="where to write (default: overwrite --base)")
+    src = ap.add_mutually_exclusive_group()
+    src.add_argument("--from-file", type=Path, help="read the submission list from a local file")
+    src.add_argument("--github-repo", default=DEFAULT_GITHUB_REPO)
+    ap.add_argument("--github-path", default=DEFAULT_GITHUB_PATH)
+    ap.add_argument("--github-ref", default=DEFAULT_GITHUB_REF,
+                    help="branch, tag or commit (recorded as the resolved commit sha)")
+    ap.add_argument("--entry-template", default=DEFAULT_ENTRY_TEMPLATE,
+                    help="repo-relative entry USD per package; {pkg} is the package folder")
+    ap.add_argument("--verify", action="store_true",
+                    help="check every package's marker and entry USD exist in the repo at the pinned sha")
+    args = ap.parse_args(argv)
+
+    if args.from_file:
+        text, source = args.from_file.read_text(), {"file": str(args.from_file)}
+    else:
+        text, source = fetch_from_github(args.github_repo, args.github_path, args.github_ref)
+    try:
+        sub = parse_submission(text)
+        doc = build_corpus(json.loads(args.base.read_text()), sub, args.entry_template, source)
+    except SubmissionError as exc:
+        raise SystemExit(f"submission list: {exc}")
+    print(f"{len(sub.packages)} packages from https://{sub.host}/datasets/{sub.repo_id} @ {sub.revision}")
+
+    if args.verify:
+        missing = verify_against_repo(sub, args.entry_template)
+        if missing:
+            print(f"{len(missing)} paths missing at {sub.revision}:", file=sys.stderr)
+            for path in missing[:50]:
+                print(f"  {path}", file=sys.stderr)
+            return 1
+        print("verified: every marker and entry USD present at the pinned sha")
+
+    out = args.out or args.base
+    out.write_text(json.dumps(doc, indent=4) + "\n")
+    print(f"wrote {out}")
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())

pipeline/upload.py

@@ -39,7 +39,7 @@ def prune(cfg: Config, manifest: Dict) -> None:
         delete_prefix(cfg, f"{base}/{stale}")
 
 
-def run(cfg: Config) -> int:
+def run(cfg: Config, update_latest: bool = True) -> int:
     if not cfg.get("ARTIFACT_S3_URI").strip():
         print(
             "ARTIFACT_S3_URI is not set — this is the pipeline's one unresolved input.\n"
@@ -63,6 +63,14 @@ def run(cfg: Config) -> int:
         return 1
     print(f"  verified {len(local)} objects")
 
+    if not update_latest:
+        # Every Space without RESTORE_RUN_ID follows latest.json, so leaving
+        # it alone publishes the run without switching any live Space to it.
+        # Pruning is skipped too: it must never race the run that is current.
+        print(f"\npublished {manifest['run_id']} — latest.json NOT updated (--no-latest); "
+              f"use RESTORE_RUN_ID={manifest['run_id']} to load it")
+        return 0
+
     # Written last, and only now: this is what marks the run as current.
     pointer = {
         "run_id": manifest["run_id"],