gw#587 serving bootstrap: self-mint boots advertise their own key and run the warmup proof

src/gen_worker/executor.py

@@ -1385,11 +1385,37 @@ class _CompileTargetRecord:
 
 @dataclass(frozen=True)
 class _CompileArtifactSelection:
-    """One immutable hub-attached cell selected before model setup."""
+    """One immutable compiled-artifact identity active on a pipeline.
+
+    ``self_mint=False``: a hub-attached (store-served) cell selected before
+    model setup — the gw#577 digest receipt governs it. ``self_mint=True``:
+    this worker's OWN boot-warmup mint (gw#587 serving bootstrap) — ref is
+    the worker's self-computed key ref and the digest is self-attested; the
+    warmup proof, not a store receipt, gates serving.
+    """
 
     path: Path
     ref: str
     snapshot_digest: str
+    self_mint: bool = False
+
+
+def _selection_for(
+    delivered: Optional["_CompileArtifactSelection"],
+    mint: Any,
+) -> Optional["_CompileArtifactSelection"]:
+    """The artifact identity a just-armed pipeline actually serves from.
+
+    A self-mint outcome WINS over the boot's delivered family selection —
+    when a delivered artifact failed to arm this object and the fleet policy
+    minted instead, recording the delivered identity would advertise bytes
+    this object does not serve (the gw#586 defect shape).
+    """
+    if mint is not None:
+        return _CompileArtifactSelection(
+            path=Path(mint.artifact), ref=str(mint.ref),
+            snapshot_digest=str(mint.snapshot_digest), self_mint=True)
+    return delivered
 
 
 @dataclass
@@ -3244,6 +3270,7 @@ async def _setup_locked(
                 # spec.compile is None.
                 arming_scope = provision.ArmingScope(
                     spec.compile, self.store._cache_dir, compile_artifact,
+                    enable=self._arming_enable,
                 )
                 with arming_scope:
                     if asyncio.iscoroutinefunction(setup):
@@ -3258,19 +3285,26 @@ async def _setup_locked(
                     slot for slot in setup_slots
                     if isinstance(inj.kwargs.get(slot), (str, Path))
                 )
+                scope_mints = arming_scope.self_mints
                 for pipe, armed in arming_scope.objects:
                     if not compile_cache.has_compile_target(pipe, spec.compile):
                         continue
                     inj.add_compile_object(pipe, self_loaded_slots)
-                    if armed and compile_selection is not None:
-                        inj.active_compile_artifacts[id(pipe)] = compile_selection
-                        if trt_engine.is_engine_ref(compile_selection.ref):
+                    selection = _selection_for(
+                        compile_selection, scope_mints.get(id(pipe)))
+                    if armed and selection is not None:
+                        inj.active_compile_artifacts[id(pipe)] = selection
+                        if trt_engine.is_engine_ref(selection.ref):
                             inj.trt_execution_before[id(pipe)] = (
                                 trt_engine.execution_count(pipe))
-            proves_inductor = bool(
-                compile_selection
-                and inj.active_compile_artifacts
-                and not trt_engine.is_engine_ref(compile_selection.ref)
+            # gw#587 serving bootstrap: the warmup PROOF gates every inductor
+            # arm — delivered (store-served) AND self-minted alike. Only the
+            # artifact SOURCE differs; a self-mint that does not actually
+            # serve its own warmup graphs must fail closed below exactly
+            # like a delivered cell that doesn't (never silent eager).
+            proves_inductor = any(
+                not trt_engine.is_engine_ref(sel.ref)
+                for sel in inj.active_compile_artifacts.values()
             )
             proof_before = {
                 id(candidate.pipeline): (
@@ -3280,6 +3314,8 @@ async def _setup_locked(
                 for candidate in inj.compile_objects
                 if proves_inductor
                 and id(candidate.pipeline) in inj.active_compile_artifacts
+                and not trt_engine.is_engine_ref(
+                    inj.active_compile_artifacts[id(candidate.pipeline)].ref)
             }
             warmup = getattr(instance, "warmup", None)
 
@@ -3407,10 +3443,16 @@ async def run_warmup() -> Tuple[int, Dict[int, set[str]]]:
                     if int(getattr(spec.compile, "lora_bucket", 0) or 0):
                         compile_cache.drop_lora_lane(pipe)
                     inj.active_compile_artifacts.pop(id(pipe), None)
-                if proven and compile_seconds >= _STORE_SERVED_COMPILE_ALARM_S:
+                if (
+                    proven
+                    and compile_selection is not None
+                    and compile_seconds >= _STORE_SERVED_COMPILE_ALARM_S
+                ):
                     # gw#587 runtime assertion: this boot ATTACHED a cell
-                    # (compile_selection is set — store-served, never a
-                    # self-mint) and at least one candidate proved a warm
+                    # (compile_selection is set — store-served; a MINTING
+                    # boot has compile_selection=None and legitimately
+                    # compiles, so it is exempt by the explicit gate above)
+                    # and at least one candidate proved a warm
                     # cache hit, yet the process burned real inductor compile
                     # wall time getting there. A delivered cell should cost
                     # ~0 here; this is the gw#586 defect class generalized —
@@ -4427,10 +4469,12 @@ async def _injection_kwargs(
                     from . import compile_cache
 
                     try:
-                        armed = await _to_thread_complete(
+                        outcome = await _to_thread_complete(
                             self._enable_compiled,
                             pipe, spec.compile, compile_artifact,
                         )
+                        armed = outcome.armed
+                        pipe_mint = outcome.self_mint
                     except compile_cache.CellSelectionBugError as exc:
                         # th#883 invariant: a SELF-REQUESTED, identity-
                         # verified cell failed to arm — by construction a
@@ -4466,14 +4510,16 @@ async def _injection_kwargs(
                             raise compile_cache.CompiledLaneUnavailableError(
                                 f"cell_selection_bug: {exc}") from exc
                         armed = False
+                        pipe_mint = None
 
                     if compile_cache.has_compile_target(pipe, spec.compile):
                         result.add_compile_object(pipe, (slot,))
-                        if armed and compile_selection is not None:
-                            result.active_compile_artifacts[id(pipe)] = compile_selection
+                        selection = _selection_for(compile_selection, pipe_mint)
+                        if armed and selection is not None:
+                            result.active_compile_artifacts[id(pipe)] = selection
                             from . import trt_engine
 
-                            if trt_engine.is_engine_ref(compile_selection.ref):
+                            if trt_engine.is_engine_ref(selection.ref):
                                 result.trt_execution_before[id(pipe)] = (
                                     trt_engine.execution_count(pipe))
                 delta = max(0, self._vram_allocated() - before)
@@ -4614,7 +4660,9 @@ def _cell_publisher(self) -> "fleet_cells.CellPublisher":
                 getattr(self._settings, "worker_image_digest", "") or ""),
         )
 
-    def _enable_compiled(self, pipe: Any, cfg: Any, artifact: Optional[Path]) -> bool:
+    def _enable_compiled(
+        self, pipe: Any, cfg: Any, artifact: Optional[Path],
+    ) -> "fleet_cells.ArmOutcome":
         """Arm the best available compiled path for a freshly loaded pipeline.
 
         gw#587: delivered cell first (unchanged semantics incl. the
@@ -4623,14 +4671,34 @@ def _enable_compiled(self, pipe: Any, cfg: Any, artifact: Optional[Path]) -> boo
         and publishes through the hub's attested gate so the next worker on
         this key is store-served. Eager fallback and the fail-closed cell
         wait are gone for reachable mints; genuine mint impossibilities
-        keep the old miss policy (plain=eager, quantized=typed refusal)."""
+        keep the old miss policy (plain=eager, quantized=typed refusal).
+
+        Returns the fleet ``ArmOutcome``; a ``self_mint`` result is recorded
+        into ``active_compile_artifacts`` exactly like a delivered cell so
+        the warmup proof runs and the target advertises the worker's own
+        key ref (th#910 self-attested dispatch fence)."""
         from . import fleet_cells
 
         return fleet_cells.enable_compiled(
             pipe, cfg, self.store._cache_dir, artifact,
             publisher=self._cell_publisher(),
         )
 
+    def _arming_enable(
+        self, pipe: Any, cfg: Any, cache_dir: Optional[Path],
+        artifact: Optional[Path],
+    ) -> "fleet_cells.ArmOutcome":
+        """ArmingScope adapter: a self-loaded pipeline's ``arm_compile()``
+        gets the same fleet policy (delivered cell first, self-mint on miss)
+        as a worker-loaded slot. ``cache_dir`` comes from the scope, which
+        the executor constructed with its own store cache dir."""
+        from . import fleet_cells
+
+        return fleet_cells.enable_compiled(
+            pipe, cfg, cache_dir, artifact,
+            publisher=self._cell_publisher(),
+        )
+
     @staticmethod
     def _vram_allocated() -> int:
         if torch is not None and torch.cuda.is_available():

src/gen_worker/fleet_cells.py

@@ -44,6 +44,7 @@
 import tempfile
 import threading
 import time
+from dataclasses import dataclass
 from pathlib import Path
 from typing import Any, Callable, Optional
 
@@ -52,6 +53,41 @@
 
 logger = logging.getLogger(__name__)
 
+
+@dataclass(frozen=True)
+class SelfMint:
+    """Identity of one successfully adopted self-minted cell.
+
+    The serving-bootstrap half of gw#587/th#910: the minting worker must
+    ADVERTISE its armed target under its own key so the hub's self-attested
+    dispatch fence (``ActiveCompileRef == KeyRef(family, requested_cell_key)``)
+    and ``active_compile_artifacts`` accounting treat the mint exactly like a
+    delivered cell — the warmup proof, not the artifact source, gates serving.
+    """
+
+    family: str
+    cell_key: str
+    ref: str  # "_system/family-<f>#<key>" — compile_cache.system_repo + key
+    snapshot_digest: str  # "blake3:<hex>" of the packed artifact (self-attested)
+    artifact: Path
+
+
+@dataclass(frozen=True)
+class ArmOutcome:
+    """Result of the fleet arming policy for one pipeline.
+
+    ``armed`` mirrors the old boolean; ``self_mint`` is set only when the
+    arm was satisfied by this worker's OWN mint (never for a delivered
+    cell), letting the executor synthesize the artifact selection it
+    records/advertises.
+    """
+
+    armed: bool
+    self_mint: Optional[SelfMint] = None
+
+    def __bool__(self) -> bool:
+        return self.armed
+
 _INTENT_TIMEOUT_S = 30
 _COMPLETE_TIMEOUT_S = 30
 
@@ -206,7 +242,7 @@ def enable_compiled(
     cache_dir: Optional[Path] = None,
     artifact: Optional[Path] = None,
     publisher: Optional[CellPublisher] = None,
-) -> bool:
+) -> ArmOutcome:
     """Fleet arming policy (gw#587): delivered cell first, self-mint on miss.
 
     Replaces the executor's bare ``provision.enable_compiled`` call. HIT
@@ -216,11 +252,16 @@ def enable_compiled(
     exits are genuine mint impossibilities (no CUDA, no C toolchain, the
     mint itself failing), where plain lanes serve eager and quantized lanes
     keep their typed refusal — exactly the cozy-local store policy.
+
+    Returns :class:`ArmOutcome`; ``self_mint`` carries the minted cell's
+    identity so the caller can record/advertise it (serving-bootstrap half
+    of th#910 — the hub's self-attested fence needs the worker to claim its
+    own key as the active compile ref).
     """
     family = str(getattr(cfg, "family", "") or "")
     try:
         if provision.enable_compiled(pipe, cfg, cache_dir, artifact):
-            return True
+            return ArmOutcome(armed=True)
         # Plain-lane miss: no cell delivered / artifact unusable. Fall
         # through to the self-mint instead of the pre-gw#587 silent eager.
     except cc.CellSelectionBugError:
@@ -287,6 +328,26 @@ def enable_compiled(
             cc.drop_lora_lane(pipe)
         return _fail_closed(pipe, "minted cell failed re-adoption")
 
+    # Advertised identity (serving-bootstrap, th#910): the worker's OWN key
+    # ref + a self-attested digest of the packed bytes. Computed BEFORE the
+    # publish thread starts — its cleanup removes the mint dir (adoption
+    # already staged the cache-dir copy, so ``artifact`` is advisory after
+    # this call returns).
+    from .convert.hub import blake3_file
+
+    key = str(meta.get("cell_key") or "").strip()
+    if not key:
+        from . import cell_key
+
+        key = cell_key.from_artifact_metadata(meta).digest
+    minted = SelfMint(
+        family=family,
+        cell_key=key,
+        ref=f"{cc.system_repo(family)}#{key}",
+        snapshot_digest="blake3:" + blake3_file(target),
+        artifact=target,
+    )
+
     # Serve first, publish behind (gw#587: publish failure never blocks the
     # request that triggered the miss). The hub's attested gate decides
     # accept/refuse; cozy-local never reaches here (no publisher).
@@ -301,10 +362,10 @@ def enable_compiled(
         logger.warning(
             "fleet-cells: SELF_MINT_WITHOUT_PUBLISH_SINK family=%s — cell "
             "stays local to this pod; the fleet store gains nothing", family)
-    return True
+    return ArmOutcome(armed=True, self_mint=minted)
 
 
-def _fail_closed(pipe: Any, reason: str) -> bool:
+def _fail_closed(pipe: Any, reason: str) -> ArmOutcome:
     """The quantized-lane policy at every exit that cannot produce a cell:
     plain lanes serve eager (never-raise miss policy), w8a8/w4a4 keep the
     typed refusal (same as the cozy-local store / pre-gw#587 production)."""
@@ -316,7 +377,7 @@ def _fail_closed(pipe: Any, reason: str) -> bool:
             f"{lane[:4].upper()} requires a compile cell and the self-mint "
             f"is unavailable ({reason})")
     logger.info("fleet-cells: serving eager (%s)", reason)
-    return False
+    return ArmOutcome(armed=False)
 
 
 def _cuda_ready() -> bool:
@@ -329,7 +390,9 @@ def _cuda_ready() -> bool:
 
 
 __all__ = [
+    "ArmOutcome",
     "CellPublishRefused",
     "CellPublisher",
+    "SelfMint",
     "enable_compiled",
 ]

src/gen_worker/models/provision.py

@@ -236,6 +236,16 @@ class _ArmingContext:
     # self-loading endpoints participate in object-scoped compile targets;
     # inferring them later from class attributes would be ambiguous.
     objects: list[tuple[Any, bool]]
+    # gw#587: the scope owner's arming policy. The executor routes the fleet
+    # policy (delivered cell first, self-mint on miss) here so an endpoint's
+    # own arm_compile() call gets the SAME behavior as a worker-loaded slot;
+    # None keeps the bare delivered-artifact policy (CLI / unit rigs). The
+    # callable may return a bool or an object with `.armed`/`.self_mint`
+    # (fleet_cells.ArmOutcome) — provision cannot import fleet_cells (cycle).
+    enable: Optional[Callable[[Any, Any, Optional[Path], Optional[Path]], Any]]
+    # id(pipe) -> self-mint identity (fleet_cells.SelfMint) for pipes the
+    # scope's policy armed from their OWN mint rather than a delivered cell.
+    self_mints: dict[int, Any]
 
 
 _ARMING_CTX: "contextvars.ContextVar[Optional[_ArmingContext]]" = contextvars.ContextVar(
@@ -253,14 +263,20 @@ class ArmingScope:
     def __init__(
         self, compile: Any, cache_dir: Optional[Path] = None,
         artifact: Optional[Path] = None,
+        enable: Optional[
+            Callable[[Any, Any, Optional[Path], Optional[Path]], Any]
+        ] = None,
     ) -> None:
         self._objects: list[tuple[Any, bool]] = []
+        self._self_mints: dict[int, Any] = {}
         self._value = (
             _ArmingContext(
                 compile=compile,
                 cache_dir=cache_dir,
                 artifact=artifact,
                 objects=self._objects,
+                enable=enable,
+                self_mints=self._self_mints,
             )
             if compile is not None else None
         )
@@ -281,6 +297,11 @@ def objects(self) -> tuple[tuple[Any, bool], ...]:
         """Exact ``(pipeline, armed)`` observations from this setup scope."""
         return tuple(self._objects)
 
+    @property
+    def self_mints(self) -> dict[int, Any]:
+        """``id(pipe) -> SelfMint`` for scope pipes armed from their own mint."""
+        return dict(self._self_mints)
+
 
 def arm_compile(pipe: Any) -> bool:
     """Arm ``@endpoint(compile=Compile(...))`` on a pipeline the endpoint
@@ -303,7 +324,12 @@ def arm_compile(pipe: Any) -> bool:
             "when @endpoint(compile=Compile(...)) is declared on that "
             "function/class."
         )
-    armed = enable_compiled(pipe, ctx.compile, ctx.cache_dir, ctx.artifact)
+    enable = ctx.enable if ctx.enable is not None else enable_compiled
+    outcome = enable(pipe, ctx.compile, ctx.cache_dir, ctx.artifact)
+    armed = bool(getattr(outcome, "armed", outcome))
+    mint = getattr(outcome, "self_mint", None)
+    if mint is not None:
+        ctx.self_mints[id(pipe)] = mint
     ctx.objects.append((pipe, armed))
     return armed