gw#599: managed-tier fill-source hierarchy (R2 -> volume -> local NVMe)

docs/environment.md

@@ -42,6 +42,7 @@ this page covers the worker itself.
 | `HF_HOME` | `hf_home` | HF cache root (also read by huggingface_hub itself) |
 | `TENSORHUB_URL` | `tensorhub_url` | standalone-CLI resolve base URL |
 | `TENSORHUB_CACHE_DIR` / `TENSORHUB_CAS_DIR` | `tensorhub_cache_dir` / `tensorhub_cas_dir` | move cache/CAS off `/tmp` (cozy local persistence); `CAS_DIR` also isolates the `cli/run.py` standalone CLI in tests |
+| `TENSORHUB_FILL_SOURCE_DIR` | `tensorhub_fill_source_dir` | th#850 managed-tier ruling: an endpoint-scoped datacenter-warm CAS mount (RunPod volume), checked before R2 on a blob miss and write-through warmed from R2. Never the CAS root — that always stays `TENSORHUB_CACHE_DIR`/local. tensorhub sets this only when a volume is attached; ismount-guarded, so a plain directory never gets mistaken for it |
 
 ## C2PA Content Credentials (Settings fields, th#714)
 

proto/worker_scheduler.proto

@@ -550,6 +550,15 @@ message ModelEvent {
   // one later causal compiled-runtime FAILED; empty on boot-attached cells and
   // ordinary model events.
   string target_incarnation_id = 19;
+  // th#850 managed-tier fill hierarchy (gw#599): bytes actually fetched over
+  // the network (R2 origin) to produce this ON_DISK transition, as opposed
+  // to bytes served from a warm local-disk or endpoint-volume cache hit. Set
+  // on every fresh materialization's ON_DISK event, including an explicit 0
+  // — pair with the DOWNLOADING events' bytes_total for this ref to read
+  // "warm boot ⇒ ~0 R2 bytes". Additive: old workers never set it, which a
+  // receiver must treat as identity unknown (same convention as
+  // snapshot_digest/residency_generation above), never as a verified zero.
+  int64 network_bytes = 20;
 }
 
 enum ModelState {

src/gen_worker/config/loader.py

@@ -42,6 +42,7 @@
     "TENSORHUB_TOKEN": "tensorhub_token",
     "TENSORHUB_CACHE_DIR": "tensorhub_cache_dir",
     "TENSORHUB_CAS_DIR": "tensorhub_cas_dir",
+    "TENSORHUB_FILL_SOURCE_DIR": "tensorhub_fill_source_dir",
     "CIVITAI_API_KEY": "civitai_api_key",
     "GEN_WORKER_C2PA_CERT_PATH": "c2pa_cert_path",
     "GEN_WORKER_C2PA_KEY_PATH": "c2pa_key_path",

src/gen_worker/config/settings.py

@@ -66,6 +66,10 @@ class Settings(msgspec.Struct, frozen=True, kw_only=True):
     # models/provision.py::resolve_local_path).
     tensorhub_cache_dir: str = ""  # TENSORHUB_CACHE_DIR
     tensorhub_cas_dir: str = ""    # TENSORHUB_CAS_DIR
+    # th#850 managed-tier ruling (gw#599): endpoint-scoped datacenter-warm
+    # fill source (RunPod network volume mount), tried before R2. Never the
+    # CAS root itself — see models/cache_paths.py::tensorhub_fill_source_dir.
+    tensorhub_fill_source_dir: str = ""  # TENSORHUB_FILL_SOURCE_DIR
 
     # Civitai provider credential (CIVITAI_API_KEY, alias CIVITAI_TOKEN).
     civitai_api_key: str = ""

src/gen_worker/executor.py

@@ -53,6 +53,7 @@
     InsufficientHostRamError,
 )
 from .input_assets import cleanup_input_assets, materialize_input_assets
+from .models import cozy_snapshot
 from .models import disk_gc
 from .models import provision
 from .models import residency as residency_mod
@@ -68,7 +69,7 @@
     next_offload_rung,
     release_unused_pinned_host_cache,
 )
-from .models.cache_paths import tensorhub_cas_dir
+from .models.cache_paths import tensorhub_cas_dir, tensorhub_fill_source_dir
 from .models.download import ensure_local, lookup_provider_for_ref
 from .models.errors import MissingSnapshotError, UrlExpiredError
 from .models.residency import Residency
@@ -541,11 +542,19 @@ def __init__(
         cache_dir: Optional[Path] = None,
         vram_budget_bytes: Optional[int] = None,
         disk_free_bytes_fn: Optional[Callable[[], int]] = None,
+        fill_source_dir: Optional[Path] = None,
     ) -> None:
         self._emit = emit
         self._hf_home = hf_home or None
         self._hf_token = hf_token or None
         self._cache_dir = cache_dir or tensorhub_cas_dir()
+        # th#850 managed-tier ruling (gw#599): endpoint-scoped datacenter-warm
+        # fill source (RunPod volume mount), consulted before R2 on a blob
+        # miss — resolved once at boot like _cache_dir; never the CAS root.
+        # Same `or` shape as _cache_dir above: an explicit path (tests) wins,
+        # otherwise resolve from env (production/tensorhub; unset -> None,
+        # the cozy-local/no-volume degenerate case).
+        self._fill_source_dir = fill_source_dir or tensorhub_fill_source_dir()
         self.residency = Residency(
             on_event=self._on_residency_event, vram_budget_bytes=vram_budget_bytes,
         )
@@ -585,6 +594,14 @@ def __init__(
         # Cold-ref waiters (th#763): ensure_local blocks here until the
         # hub's re-minted DOWNLOAD banks a snapshot for the ref.
         self._snapshot_waiters: Dict[str, asyncio.Event] = {}
+        # th#850 managed-tier ruling (gw#599): network_bytes for the NEXT
+        # ON_DISK transition of this ref, handed off to
+        # _on_residency_event so the one authoritative wire event Residency
+        # emits carries it — set immediately before track_disk(), consumed
+        # (popped) by _on_residency_event if it fires, cleared defensively
+        # otherwise. Avoids a second, redundant ON_DISK event and avoids
+        # widening EventFn's arity (Residency has other direct callers).
+        self._pending_network_bytes: Dict[str, int] = {}
 
     def _default_disk_free(self) -> int:
         p = Path(self._cache_dir)
@@ -625,6 +642,9 @@ def _on_residency_event(
                 )
                 if identity[0]:
                     self._resident_identities[ref] = identity
+            pending_network_bytes = self._pending_network_bytes.pop(ref, None)
+            if pending_network_bytes is not None:
+                kw["network_bytes"] = int(pending_network_bytes)
         else:
             identity = self.resident_identity(ref)
         coro = self._event(ref, pb_state, identity=identity, **kw)
@@ -965,6 +985,10 @@ def _gc_candidates(
         keep: Tuple[str, ...],
         keep_rank: Dict[str, int],
     ) -> List[str]:
+        """The evictable SET for one gc_disk pass: hard invariants only
+        (never exclude/in-use — no policy ever overrides these), plus this
+        pass's keep-membership/grace filter. Ordering within that set is a
+        separate seam, see ``_disk_eviction_order``."""
         now = time.time()
         out: List[Tuple[float, str]] = []
         for ref in self.residency.refs_in(residency_mod.Tier.DISK):
@@ -976,11 +1000,28 @@ def _gc_candidates(
             if honor_grace and (now - last) < _DISK_GC_GRACE_S:
                 continue
             out.append((last, ref))
+        return self._disk_eviction_order(out, include_keep, keep_rank)
+
+    # th#850 managed-tier ruling (gw#599): the eviction POLICY (ranking one
+    # pass's evictable set) is a distinct seam from the evictable SET itself
+    # (``_gc_candidates`` above, which owns the hard never-evict invariants).
+    # Default is the LRU-oldest-first/keep-priority-escape-hatch ordering
+    # this store has always used — an instance may swap this attribute for a
+    # scheduler-intent-aware policy without touching gc_disk's free-space
+    # loop or the invariant filter. Building that policy is a follow-on
+    # (Paul ruled the seam only here); the default below is exactly today's
+    # behavior, byte-for-byte.
+    @staticmethod
+    def _default_disk_eviction_order(
+        entries: List[Tuple[float, str]], include_keep: bool, keep_rank: Dict[str, int],
+    ) -> List[str]:
         if include_keep:
-            out.sort(key=lambda item: (-keep_rank[item[1]], item[0], item[1]))
+            ordered = sorted(entries, key=lambda item: (-keep_rank[item[1]], item[0], item[1]))
         else:
-            out.sort()
-        return [r for _, r in out]
+            ordered = sorted(entries)
+        return [ref for _, ref in ordered]
+
+    _disk_eviction_order = _default_disk_eviction_order
 
     def _evict_disk_ref(self, ref: str) -> None:
         path = self.residency.local_path(ref) or self._index.path(ref)
@@ -1153,6 +1194,14 @@ def complete(path: Path) -> _MaterializedLocal:
                     operation_identity,
                 )
             last_progress = 0.0
+            # th#850 managed-tier ruling (gw#599): opened before _progress so
+            # its DOWNLOADING ticks can read the running total, and entered
+            # once for the whole retry loop so it accumulates across
+            # attempts. The hub (tensorhub th#850/PR#493) reads network_bytes
+            # off the DOWNLOADING events' running value (mirrors
+            # bytes_done/bytes_total), not just the terminal ON_DISK one —
+            # both must carry it for the wire contract to actually work.
+            net_scope = cozy_snapshot.NetworkBytesScope()
 
             def _progress(done: int, total: Optional[int]) -> None:
                 nonlocal last_progress
@@ -1164,7 +1213,8 @@ def _progress(done: int, total: Optional[int]) -> None:
                 asyncio.run_coroutine_threadsafe(
                     self._event(ref, pb.MODEL_STATE_DOWNLOADING,
                                 identity=operation_identity,
-                                bytes_done=int(done), bytes_total=int(total or 0)),
+                                bytes_done=int(done), bytes_total=int(total or 0),
+                                network_bytes=net_scope.network_bytes),
                     self._loop,
                 )
 
@@ -1177,17 +1227,19 @@ def _progress(done: int, total: Optional[int]) -> None:
                     resolved = None
                     if snapshot is not None and snapshot.digest:
                         resolved = _snapshot_to_resolved(snapshot)
-                    path = await ensure_local(
-                        ref,
-                        provider=getattr(binding, "source", None),
-                        snapshot=resolved,
-                        cache_dir=self._cache_dir,
-                        hf_home=self._hf_home,
-                        hf_token=self._hf_token,
-                        allow_patterns=tuple(getattr(binding, "files", ()) or ()),
-                        components=tuple(getattr(binding, "components", ()) or ()),
-                        progress=_progress,
-                    )
+                    with net_scope:
+                        path = await ensure_local(
+                            ref,
+                            provider=getattr(binding, "source", None),
+                            snapshot=resolved,
+                            cache_dir=self._cache_dir,
+                            hf_home=self._hf_home,
+                            hf_token=self._hf_token,
+                            allow_patterns=tuple(getattr(binding, "files", ()) or ()),
+                            components=tuple(getattr(binding, "components", ()) or ()),
+                            progress=_progress,
+                            fill_source_dir=self._fill_source_dir,
+                        )
                     tier_before = self.residency.tier(ref)
                     with self._identity_lock:
                         identity_changed = (
@@ -1198,13 +1250,25 @@ def _progress(done: int, total: Optional[int]) -> None:
                             self._disk_identities[ref] = operation_identity
                             if tier_before in (None, residency_mod.Tier.DISK):
                                 self._resident_identities[ref] = operation_identity
+                    # th#850 managed-tier ruling (gw#599): handed off to
+                    # _on_residency_event so the ON_DISK event Residency
+                    # emits for a genuinely fresh registration carries the
+                    # bytes this materialization fetched over the network
+                    # (0 included — pairs with the DOWNLOADING events'
+                    # bytes_total for the "warm boot ⇒ ~0 R2 bytes" signal).
+                    self._pending_network_bytes[ref] = net_scope.network_bytes
                     self.residency.track_disk(ref, path)
+                    self._pending_network_bytes.pop(ref, None)  # defensive: unconsumed if no event fired
                     if tier_before is residency_mod.Tier.DISK and identity_changed:
-                        # Residency suppresses same-tier event spam. A digest
-                        # move is nevertheless a semantic ON_DISK transition.
+                        # Residency suppresses same-tier event spam (track_disk
+                        # above did not consume the pending value above). A
+                        # digest move is nevertheless a semantic ON_DISK
+                        # transition — carries network_bytes directly since
+                        # this is our own explicit event, not Residency's.
                         await self._event(
                             ref, pb.MODEL_STATE_ON_DISK,
                             identity=operation_identity,
+                            network_bytes=net_scope.network_bytes,
                         )
                     # tree_bytes stats every file — off-loop (gw#407: no
                     # multi-GB directory walks on the event loop).

src/gen_worker/models/cache_paths.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+import os
 from pathlib import Path
 
 from ..config import get_settings
@@ -13,14 +14,14 @@ def tensorhub_cache_dir() -> Path:
 
     Honors the ``TENSORHUB_CACHE_DIR`` environment variable when set. This is
     the ONE knob for where the CAS lives: the cozy-local runner points it at
-    a persistent ``~/.cache/tensorhub`` (weights survive reboots); tensorhub
-    points it at a mounted RunPod endpoint volume (`/tensorhub-endpoint-cache`)
-    when the pod has one attached, so the ordinary R2 prefetch that
-    materializes blobs here on the first boot warms the volume for later
-    same-endpoint pods too (th#850) — no separate shared/read-through tier.
-    Falls back to the ``/tmp`` default when unset. The CAS implementation
-    itself is deliberately backend-agnostic: nothing branches on what's
-    mounted here.
+    a persistent ``~/.cache/tensorhub`` (weights survive reboots). The CAS
+    root ALWAYS stays on local/pod-local storage — a managed, bounded LRU
+    tier (th#850 managed-tier ruling, gw#599: supersedes the earlier
+    CAS-root-on-volume shape). A mounted RunPod endpoint volume, when
+    attached, is a FILL SOURCE consulted before R2 (see
+    ``tensorhub_fill_source_dir``) — it is never the CAS root itself. Falls
+    back to the ``/tmp`` default when unset. The CAS implementation itself is
+    deliberately backend-agnostic: nothing branches on what's mounted here.
     """
     env = get_settings().tensorhub_cache_dir.strip()
     if env:
@@ -29,5 +30,28 @@ def tensorhub_cache_dir() -> Path:
 
 
 def tensorhub_cas_dir() -> Path:
-    """Worker CAS root: <TENSORHUB_CACHE_DIR>/cas."""
+    """Worker CAS root: <TENSORHUB_CACHE_DIR>/cas. Always local/pod storage."""
     return tensorhub_cache_dir() / "cas"
+
+
+def tensorhub_fill_source_dir() -> Path | None:
+    """Endpoint-scoped datacenter-warm fill source (th#850 managed-tier
+    ruling), or ``None`` when no volume is attached.
+
+    Honors ``TENSORHUB_FILL_SOURCE_DIR``, set by tensorhub only when this
+    pod's endpoint has a RunPod network volume attached. Guarded by
+    ``os.path.ismount`` — a plain directory baked into the image or left on
+    the container disk must never be mistaken for the real per-endpoint
+    volume (same defensive pattern the removed ``endpoint_shared_cas_dir``
+    used). This is FILL SOURCE #1 in the CAS layer's fetch order (volume,
+    then R2); it is never the CAS root. cozy-local and any pod without a
+    volume leave this unset, which is the degenerate case: fetch goes
+    straight to R2, byte-identical to pre-th#850 behavior.
+    """
+    env = get_settings().tensorhub_fill_source_dir.strip()
+    if not env:
+        return None
+    path = Path(env).expanduser()
+    if not os.path.ismount(path):
+        return None
+    return path