Studio: bring back the Resume button for past training runs - #11301

diff --git a/studio/backend/core/training/provenance.py b/studio/backend/core/training/provenance.py
index 6781c3795..99ef7d1ac 100644
--- a/studio/backend/core/training/provenance.py
+++ b/studio/backend/core/training/provenance.py
@@ -195,7 +195,7 @@ def _snapshot_declares_quantization(snapshot: Path) -> bool:
 
 
 def _resolved_model_snapshot_file(snapshot: Path, path: Path) -> Optional[Path]:
-    from hub.utils.hf_cache_state import same_existing_path
+    from hub.utils.hf_cache_state import same_existing_path, trusted_blob_roots
 
     try:
         snapshot = snapshot.resolve(strict = True)
@@ -208,8 +208,7 @@ def _resolved_model_snapshot_file(snapshot: Path, path: Path) -> Optional[Path]:
         return None
     if not resolved.is_file() or not (
         resolved.is_relative_to(snapshot)
-        or resolved.is_relative_to(repo_dir / "blobs")
-        or resolved.is_relative_to(repo_dir.parent / "blobs")
+        or any(resolved.is_relative_to(root) for root in trusted_blob_roots(repo_dir))
     ):
         return None
     try:
diff --git a/studio/backend/hub/utils/dataset_cache.py b/studio/backend/hub/utils/dataset_cache.py
index 03bb77559..419f63055 100644
--- a/studio/backend/hub/utils/dataset_cache.py
+++ b/studio/backend/hub/utils/dataset_cache.py
@@ -20,6 +20,7 @@ from hub.utils.hf_cache_state import (
     iter_repo_cache_dirs,
     ref_snapshot_dir,
     same_existing_path,
+    trusted_blob_roots,
     validated_repo_cache_path,
 )
 from utils.paths.path_utils import drop_appledouble_metadata, is_appledouble_metadata
@@ -279,8 +280,7 @@ def resolved_dataset_snapshot_file(snapshot: str | Path, source_path: str) -> Op
         return None
     if not resolved.is_file() or not (
         resolved.is_relative_to(snapshot_path)
-        or resolved.is_relative_to(repo_dir / "blobs")
-        or resolved.is_relative_to(repo_dir.parent / "blobs")
+        or any(resolved.is_relative_to(root) for root in trusted_blob_roots(repo_dir))
     ):
         return None
     try:
diff --git a/studio/backend/hub/utils/hf_cache_state.py b/studio/backend/hub/utils/hf_cache_state.py
index da97214de..9f1607571 100644
--- a/studio/backend/hub/utils/hf_cache_state.py
+++ b/studio/backend/hub/utils/hf_cache_state.py
@@ -142,6 +142,59 @@ def same_existing_path(first: Path, second: Path) -> bool:
         return False
 
 
+SHARED_BLOBS_MARKER_NAME = ".huggingface-shared-blobs"
+
+
+def _resolved_existing_dir(path: Path) -> Optional[Path]:
+    try:
+        return path.resolve(strict = True) if path.is_dir() else None
+    except (OSError, RuntimeError, ValueError):
+        return None
+
+
+def _is_hub_shared_blobs_dir(path: Path) -> bool:
+    """Is this huggingface_hub's own cache-wide shared blob store, not just a folder called ``blobs``?
+
+    Read from upstream rather than reimplemented: hub validates an ownership marker AND its
+    layout version, and a later hub may bump that version, so asking the installed hub keeps
+    us in step with the cache it actually writes. The literal marker check is the fallback for
+    a hub too old to export the helper -- such a hub also never creates the store, so the
+    fallback answers False and the caller behaves exactly as it did before 1.32.
+    """
+    try:
+        from huggingface_hub.utils._shared_blobs import is_shared_blobs_dir
+        return bool(is_shared_blobs_dir(path))
+    except Exception:
+        pass
+    try:
+        return path.is_dir() and (path / SHARED_BLOBS_MARKER_NAME).is_file()
+    except (OSError, ValueError):
+        return False
+
+
+def trusted_blob_roots(repo_dir: Path) -> tuple[Path, ...]:
+    """Resolved directories a file inside ``repo_dir``'s snapshot may legitimately resolve into.
+
+    The repo's own ``blobs``, plus huggingface_hub 1.32's cache-wide shared Xet store at
+    ``<cache_root>/blobs``: 1.32 turned each repo's ``blobs/<etag>`` into a symlink into that
+    store, so a weight file resolves outside its repo folder without leaving the cache.
+
+    Both roots are RESOLVED, because callers compare them against a fully resolved candidate.
+    Comparing against a literal path silently rejects the very files it is meant to admit when
+    a ``blobs`` leaf is a symlink, which is how a big shared store ends up on another volume.
+    """
+    roots: list[Path] = []
+    own = _resolved_existing_dir(repo_dir / "blobs")
+    if own is not None:
+        roots.append(own)
+    shared = repo_dir.parent / "blobs"
+    if _is_hub_shared_blobs_dir(shared):
+        resolved_shared = _resolved_existing_dir(shared)
+        if resolved_shared is not None and resolved_shared not in roots:
+            roots.append(resolved_shared)
+    return tuple(roots)
+
+
 def hf_cache_root(
     *,
     create: bool = False,
diff --git a/studio/backend/routes/models.py b/studio/backend/routes/models.py
index 37f86499b..60fe0d263 100644
--- a/studio/backend/routes/models.py
+++ b/studio/backend/routes/models.py
@@ -2081,14 +2081,12 @@ def _get_snapshot_model_size_bytes(snapshot_path: str) -> Optional[int]:
         repo_dir = snapshots_dir.parent.resolve(strict = True)
         if not snapshot.is_dir() or snapshots_dir.name != "snapshots" or not repo_dir.is_dir():
             return None
-        blobs_dir = repo_dir / "blobs"
-        resolved_blobs_dir = blobs_dir.resolve(strict = True) if blobs_dir.is_dir() else None
-        # hub 1.x keeps one content-addressed blob store per cache root and links each repo's
-        # blobs into it, so a weight file resolves outside the repo without leaving the cache.
-        shared_blobs_dir = repo_dir.parent / "blobs"
-        resolved_shared_blobs_dir = (
-            shared_blobs_dir.resolve(strict = True) if shared_blobs_dir.is_dir() else None
-        )
+        # The repo's own blobs, plus hub 1.32's cache-wide shared store: it links each repo's
+        # blobs into one content-addressed folder per cache root, so a weight file resolves
+        # outside the repo without leaving the cache. Same roots the attestation check trusts.
+        from hub.utils.hf_cache_state import trusted_blob_roots
+
+        blob_roots = trusted_blob_roots(repo_dir)
     except (OSError, RuntimeError, ValueError):
         return None
 
@@ -2114,8 +2112,7 @@ def _get_snapshot_model_size_bytes(snapshot_path: str) -> Optional[int]:
                     if not candidate.is_file():
                         continue
                     if not candidate.is_relative_to(snapshot) and not any(
-                        blob_root is not None and candidate.is_relative_to(blob_root)
-                        for blob_root in (resolved_blobs_dir, resolved_shared_blobs_dir)
+                        candidate.is_relative_to(blob_root) for blob_root in blob_roots
                     ):
                         continue
                     total += candidate.stat().st_size
diff --git a/studio/backend/tests/test_model_size_shared_blob_store.py b/studio/backend/tests/test_model_size_shared_blob_store.py
index 734966eb6..b5d507e9b 100644
--- a/studio/backend/tests/test_model_size_shared_blob_store.py
+++ b/studio/backend/tests/test_model_size_shared_blob_store.py
@@ -39,8 +39,16 @@ def _link_through_repo_blob(snapshot: Path, name: str, target: Path) -> None:
     (snapshot / name).symlink_to(os.path.relpath(repo_blob, snapshot))
 
 
+def _mark_shared_store(cache_root: Path) -> None:
+    """The ownership marker hub writes at the root of a store it created."""
+    store = cache_root / "blobs"
+    store.mkdir(parents = True, exist_ok = True)
+    (store / ".huggingface-shared-blobs").write_text("1\n")
+
+
 def test_model_size_counts_weights_in_the_hub_shared_blob_store(tmp_path):
     snapshot = _snapshot(tmp_path, "org/model")
+    _mark_shared_store(tmp_path)
     sha = "8788269b" * 8
     shared = tmp_path / "blobs" / sha[:2] / sha
     shared.parent.mkdir(parents = True)
@@ -50,6 +58,38 @@ def test_model_size_counts_weights_in_the_hub_shared_blob_store(tmp_path):
     assert models_route._get_snapshot_model_size_bytes(str(snapshot)) == len(WEIGHTS)
 
 
+def test_model_size_ignores_a_weight_under_an_unmarked_blobs_dir(tmp_path):
+    """Sizing trusts the same roots attestation does, so a bare ``blobs`` folder is not one."""
+    snapshot = _snapshot(tmp_path, "org/model")
+    sha = "8788269b" * 8
+    unmarked = tmp_path / "blobs" / sha[:2] / sha
+    unmarked.parent.mkdir(parents = True)
+    unmarked.write_bytes(WEIGHTS)
+    _link_through_repo_blob(snapshot, "model.safetensors", unmarked)
+
+    assert models_route._get_snapshot_model_size_bytes(str(snapshot)) is None
+
+
+def test_model_size_ignores_a_weight_in_a_symlinked_shared_store(tmp_path):
+    """Sizing and attestation answer the same question about a symlinked ``blobs`` leaf.
+
+    Before, sizing accepted it and attestation did not, so the model reported a size while its
+    run stayed unresumable and said the revision was not attested.
+    """
+    snapshot = _snapshot(tmp_path, "org/model")
+    elsewhere = tmp_path / "another-volume"
+    elsewhere.mkdir()
+    (tmp_path / "blobs").symlink_to(elsewhere)
+    (elsewhere / ".huggingface-shared-blobs").write_text("1\n")
+    sha = "8788269b" * 8
+    shared = elsewhere / sha[:2] / sha
+    shared.parent.mkdir(parents = True)
+    shared.write_bytes(WEIGHTS)
+    _link_through_repo_blob(snapshot, "model.safetensors", shared)
+
+    assert models_route._get_snapshot_model_size_bytes(str(snapshot)) is None
+
+
 def test_model_size_still_counts_a_weight_in_the_repos_own_blobs(tmp_path):
     snapshot = _snapshot(tmp_path, "org/model")
     repo_blob = snapshot.parent.parent / "blobs" / ("a1b2c3d4" * 8)
diff --git a/studio/backend/tests/test_training_provenance.py b/studio/backend/tests/test_training_provenance.py
index 776e586dc..803239249 100644
--- a/studio/backend/tests/test_training_provenance.py
+++ b/studio/backend/tests/test_training_provenance.py
@@ -635,8 +635,24 @@ def test_exact_model_snapshot_accepts_own_blob_symlink(tmp_path):
     assert exact_model_snapshot_path(str(snapshot), "org/model") == str(snapshot.resolve())
 
 
-def _shared_store_blob_symlink(repo: Path, link: Path, payload: bytes) -> Path:
+def _mark_shared_store(cache_root: Path) -> Path:
+    """Write the ownership marker hub puts at the root of a store it created.
+
+    Without it the directory is just a folder named ``blobs``, and a fixture that omits it
+    proves acceptance of any such folder rather than of hub's store.
+    """
+    store = cache_root / "blobs"
+    store.mkdir(parents = True, exist_ok = True)
+    (store / ".huggingface-shared-blobs").write_text("1\n")
+    return store
+
+
+def _shared_store_blob_symlink(
+    repo: Path, link: Path, payload: bytes, *, marked: bool = True
+) -> Path:
     sha = "c791637d" * 8
+    if marked:
+        _mark_shared_store(repo.parent)
     shared = repo.parent / "blobs" / sha[:2] / sha
     shared.parent.mkdir(parents = True, exist_ok = True)
     shared.write_bytes(payload)
@@ -655,6 +671,37 @@ def test_exact_model_snapshot_accepts_hub_shared_blob_store(tmp_path):
     assert exact_model_snapshot_path(str(snapshot), "org/model") == str(snapshot.resolve())
 
 
+def test_exact_model_snapshot_rejects_unmarked_blobs_dir(tmp_path):
+    """A folder called ``blobs`` beside the repo is not hub's shared store.
+
+    Only hub creates that store, and it marks what it created. Trusting the name alone would
+    make every readable file under it attestable, which is a wider set than the cache ever holds.
+    """
+    snapshot = _model_snapshot(tmp_path, "org/model", "unmarked", weights = False)
+    _shared_store_blob_symlink(
+        snapshot.parent.parent, snapshot / "model.safetensors", b"weights", marked = False
+    )
+
+    assert exact_model_snapshot_path(str(snapshot), "org/model") is None
+
+
+def test_exact_model_snapshot_rejects_symlinked_shared_store(tmp_path):
+    """A ``blobs`` leaf that is itself a symlink is not a store hub owns.
+
+    ``is_shared_blobs_dir`` lstats the leaf, so hub never adopts one and never publishes into
+    it. Sizing asks the same question, so the two cannot disagree and leave a model whose size
+    displays but whose run will not resume.
+    """
+    snapshot = _model_snapshot(tmp_path, "org/model", "relocated", weights = False)
+    elsewhere = tmp_path / "another-volume"
+    elsewhere.mkdir()
+    (tmp_path / "blobs").symlink_to(elsewhere)
+    (elsewhere / ".huggingface-shared-blobs").write_text("1\n")
+    _shared_store_blob_symlink(snapshot.parent.parent, snapshot / "model.safetensors", b"weights")
+
+    assert exact_model_snapshot_path(str(snapshot), "org/model") is None
+
+
 @pytest.mark.parametrize("target", ["outside-cache", "other-repo-blobs"])
 def test_exact_model_snapshot_rejects_blob_symlink_escaping_repo(
     tmp_path, tmp_path_factory, target
@@ -891,6 +938,31 @@ def test_loaded_hub_dataset_accepts_hub_shared_blob_store(tmp_path):
     assert exact_dataset_snapshot_path(str(snapshot), "org/dataset") == str(snapshot.resolve())
 
 
+@pytest.mark.parametrize("target", ["outside-cache", "other-repo-blobs", "unmarked-blobs"])
+def test_loaded_hub_dataset_rejects_blob_symlink_escaping_repo(
+    tmp_path, tmp_path_factory, target
+):
+    """The dataset predicate was widened exactly as the model one was, so it is bounded the same."""
+    repo = tmp_path / "datasets--org--dataset"
+    snapshot = repo / "snapshots" / "dataset-commit"
+    snapshot.mkdir(parents = True)
+    if target == "unmarked-blobs":
+        _shared_store_blob_symlink(repo, snapshot / "train.parquet", b"dataset", marked = False)
+    else:
+        if target == "outside-cache":
+            escaped = tmp_path_factory.mktemp("elsewhere") / "blobs" / "c7" / "train.parquet"
+        else:
+            escaped = tmp_path / "datasets--org--other" / "blobs" / "train.parquet"
+        escaped.parent.mkdir(parents = True)
+        escaped.write_bytes(b"dataset")
+        repo_blob = repo / "blobs" / "dataset-blob"
+        repo_blob.parent.mkdir(parents = True)
+        repo_blob.symlink_to(escaped)
+        (snapshot / "train.parquet").symlink_to(os.path.relpath(repo_blob, snapshot))
+
+    assert exact_dataset_snapshot_path(str(snapshot), "org/dataset") is None
+
+
 def test_loaded_hub_dataset_rejects_local_source_symlink_outside_repo(tmp_path):
     snapshot = _shared_setup_1(tmp_path)
     external = tmp_path / "external.parquet"

diff --git a/studio/backend/hub/utils/hf_cache_state.py b/studio/backend/hub/utils/hf_cache_state.py
index 9f1607571..7ad778787 100644
--- a/studio/backend/hub/utils/hf_cache_state.py
+++ b/studio/backend/hub/utils/hf_cache_state.py
@@ -143,11 +143,22 @@ def same_existing_path(first: Path, second: Path) -> bool:
 
 
 SHARED_BLOBS_MARKER_NAME = ".huggingface-shared-blobs"
+_SHARED_BLOBS_LAYOUT_RE = re.compile(r"\A[0-9]+\n\Z")
 
 
-def _resolved_existing_dir(path: Path) -> Optional[Path]:
+def _resolved_real_dir(path: Path) -> Optional[Path]:
+    """Resolve ``path``, but only if the leaf itself is a real directory.
+
+    ``is_dir()`` follows symlinks and ``resolve()`` then returns wherever the link points, so
+    resolving first and asking questions afterwards would make a ``blobs`` symlink into a
+    trust root anchored outside the cache. Callers compare a fully resolved candidate against
+    what this returns, so that would admit externally mutable bytes as an exact snapshot.
+    ``lstat`` is the difference: it reports the link, not its target.
+    """
     try:
-        return path.resolve(strict = True) if path.is_dir() else None
+        if not stat_module.S_ISDIR(path.lstat().st_mode):
+            return None
+        return path.resolve(strict = True)
     except (OSError, RuntimeError, ValueError):
         return None
 
@@ -157,17 +168,24 @@ def _is_hub_shared_blobs_dir(path: Path) -> bool:
 
     Read from upstream rather than reimplemented: hub validates an ownership marker AND its
     layout version, and a later hub may bump that version, so asking the installed hub keeps
-    us in step with the cache it actually writes. The literal marker check is the fallback for
-    a hub too old to export the helper -- such a hub also never creates the store, so the
-    fallback answers False and the caller behaves exactly as it did before 1.32.
+    us in step with the cache it actually writes. The fallback is for a hub too old to export
+    the helper, which also never creates the store, so it should and does answer False for
+    everything a real cache contains; it mirrors upstream's shape (real directory, regular
+    marker file, a layout version line) rather than trusting a filename, since otherwise a
+    hand-made marker would buy trust upstream itself would refuse.
     """
     try:
         from huggingface_hub.utils._shared_blobs import is_shared_blobs_dir
         return bool(is_shared_blobs_dir(path))
     except Exception:
         pass
+    marker = path / SHARED_BLOBS_MARKER_NAME
     try:
-        return path.is_dir() and (path / SHARED_BLOBS_MARKER_NAME).is_file()
+        if not stat_module.S_ISDIR(path.lstat().st_mode):
+            return False
+        if not stat_module.S_ISREG(marker.lstat().st_mode):
+            return False
+        return _SHARED_BLOBS_LAYOUT_RE.fullmatch(marker.read_text()) is not None
     except (OSError, ValueError):
         return False
 
@@ -179,17 +197,17 @@ def trusted_blob_roots(repo_dir: Path) -> tuple[Path, ...]:
     ``<cache_root>/blobs``: 1.32 turned each repo's ``blobs/<etag>`` into a symlink into that
     store, so a weight file resolves outside its repo folder without leaving the cache.
 
-    Both roots are RESOLVED, because callers compare them against a fully resolved candidate.
-    Comparing against a literal path silently rejects the very files it is meant to admit when
-    a ``blobs`` leaf is a symlink, which is how a big shared store ends up on another volume.
+    Both roots are RESOLVED, because callers compare them against a fully resolved candidate,
+    and a ``blobs`` leaf that is itself a symlink is not a root at all: hub refuses to adopt
+    one as its store, and honouring one here would anchor trust wherever the link points.
     """
     roots: list[Path] = []
-    own = _resolved_existing_dir(repo_dir / "blobs")
+    own = _resolved_real_dir(repo_dir / "blobs")
     if own is not None:
         roots.append(own)
     shared = repo_dir.parent / "blobs"
     if _is_hub_shared_blobs_dir(shared):
-        resolved_shared = _resolved_existing_dir(shared)
+        resolved_shared = _resolved_real_dir(shared)
         if resolved_shared is not None and resolved_shared not in roots:
             roots.append(resolved_shared)
     return tuple(roots)
diff --git a/studio/backend/tests/test_training_provenance.py b/studio/backend/tests/test_training_provenance.py
index 803239249..c4fe91cdf 100644
--- a/studio/backend/tests/test_training_provenance.py
+++ b/studio/backend/tests/test_training_provenance.py
@@ -685,6 +685,50 @@ def test_exact_model_snapshot_rejects_unmarked_blobs_dir(tmp_path):
     assert exact_model_snapshot_path(str(snapshot), "org/model") is None
 
 
+def test_exact_model_snapshot_rejects_symlinked_repo_blobs_dir(tmp_path):
+    """A repo's own ``blobs`` that is a symlink anchors trust wherever it points.
+
+    The containment test compares against a resolved root, so resolving a symlinked leaf
+    would make every file under its target an exact snapshot, including externally mutable
+    bytes that are not in the cache at all.
+    """
+    snapshot = _model_snapshot(tmp_path, "org/model", "linked-blobs", weights = False)
+    repo = snapshot.parent.parent
+    outside = tmp_path / "not-the-cache"
+    outside.mkdir()
+    (repo / "blobs").symlink_to(outside)
+    payload = outside / "etag"
+    payload.write_bytes(b"weights")
+    (snapshot / "model.safetensors").symlink_to(os.path.relpath(payload, snapshot))
+
+    assert exact_model_snapshot_path(str(snapshot), "org/model") is None
+
+
+def test_exact_model_snapshot_rejects_hand_made_shared_store_marker(tmp_path, monkeypatch):
+    """With no hub helper to ask, a bare marker filename must not buy trust.
+
+    The fallback only runs on a hub old enough to lack the store entirely, so anything it
+    sees is hand made; upstream requires a regular marker holding a layout version, and so
+    does this.
+    """
+    import sys
+
+    # None in sys.modules makes the import raise, which is the state a pre-1.32 hub presents.
+    monkeypatch.setitem(sys.modules, "huggingface_hub.utils._shared_blobs", None)
+    snapshot = _model_snapshot(tmp_path, "org/model", "hand-made", weights = False)
+    store = tmp_path / "blobs"
+    store.mkdir()
+    (store / ".huggingface-shared-blobs").write_text("not a layout version")
+    payload = store / "payload"
+    payload.write_bytes(b"weights")
+    repo_blob = snapshot.parent.parent / "blobs" / "etag"
+    repo_blob.parent.mkdir(exist_ok = True)
+    repo_blob.symlink_to(os.path.relpath(payload, repo_blob.parent))
+    (snapshot / "model.safetensors").symlink_to(os.path.relpath(repo_blob, snapshot))
+
+    assert exact_model_snapshot_path(str(snapshot), "org/model") is None
+
+
 def test_exact_model_snapshot_rejects_symlinked_shared_store(tmp_path):
     """A ``blobs`` leaf that is itself a symlink is not a store hub owns.