diff --git a/src/gen_worker/compile_cache.py b/src/gen_worker/compile_cache.py index 01a95881..d052e12e 100644 --- a/src/gen_worker/compile_cache.py +++ b/src/gen_worker/compile_cache.py @@ -2042,6 +2042,86 @@ def mint_artifact( return meta +def begin_fleet_mint(pipe: Any, cfg: Any, capture: Path) -> None: + """Arm ``pipe`` for a fleet self-mint capture (gw#587 CORRECT FIX). + + Points inductor/triton at a fresh ``capture`` dir and enables the + declared compile targets in cold-allowed, GUARDED mode — WITHOUT + running any synthetic warm call (the old ``mint_artifact``/ + ``_warm_call`` producer-style recipe, gw#586's defect class + resurfacing inside self-mint, gw#587's root cause). + + The caller's real serving warmup — the executor's own warmup-proof + window, running the endpoint's own code (the two-stage/conditioned + call LTX and its siblings actually make) — performs the ONLY compile + this mint will ever see. Capturing that exact execution instead of a + second, separately-shaped call is what makes the published artifact + byte-derived from the same execution the proof observed: there is no + other code path that could re-create serving's call shape, so the + mint can never diverge from what it actually serves. + + Raises if no compile target resolves on ``pipe`` (nothing to prove or + publish); the caller's miss policy applies exactly as it did for a + failed :func:`mint_artifact` call. + """ + capture_env(capture) + if not apply(pipe, cfg, cache_ready=False, guard=True, allow_cold=True): + raise RuntimeError(f"no compile targets resolved on {type(pipe).__name__}") + + +def finish_fleet_mint( + pipe: Any, cfg: Any, family: str, target: Path, capture: Path, +) -> Dict[str, Any]: + """Pack the capture dir a PASSED warmup proof just populated. + + Callers must invoke this ONLY after the executor's warmup proof has + confirmed the real serving call exercised ``pipe``'s attached compile + targets (a successful compiled call recorded — never before: packing + ahead of the proof would reopen the publish-before-proof window + gw#587 closes, and packing an unexercised/failed capture would + publish bytes nothing ever proved served). + + Unlike :func:`mint_artifact`, this never compiles anything itself — + the compile already happened, inside the proof window, driven by the + endpoint's own warmup. This function only packages what that warmup + produced. + """ + captured = [p for p in (capture / "inductor").rglob("*") if p.is_file()] + if not captured: + raise RuntimeError( + "self-mint proof passed but captured nothing under " + "TORCHINDUCTOR_CACHE_DIR — was inductor already latched to " + "another dir in this process?" + ) + from .models.loading import pipeline_weight_lane + from .models.memory import low_vram_mode + + # gw#564: record the execution contract exactly like the production + # build — w8a8 cells are contract_drift-gated on the graph signature and + # weight-lane manifest, so a mint without them can never re-adopt. Both + # are STATIC (module structure + declared shapes/targets), computed the + # same way whether sampled before or after the real compile. + graph_signature, weight_contract = execution_contract(pipe, cfg) + meta = artifact_metadata( + family=family, + source_ref="self-mint", + shapes=cfg.shapes, + targets=cfg.targets, + guidance_scales=getattr(cfg, "guidance_scales", ()), + low_vram_mode=low_vram_mode(pipe), + compile_mode="regional" if getattr(cfg, "regional", False) else "whole", + weight_lane=pipeline_weight_lane(pipe), + lora_bucket=int(getattr(cfg, "lora_bucket", 0) or 0), + graph_signature=graph_signature, + weight_contract=weight_contract, + ) + tmp = target.with_suffix(".part") + target.parent.mkdir(parents=True, exist_ok=True) + pack(capture, tmp, meta) + os.replace(tmp, target) + return meta + + __all__ = [ "ARTIFACT_FORMAT", "AdoptError", @@ -2051,6 +2131,7 @@ def mint_artifact( "apply", "apply_lora_lane", "artifact_metadata", + "begin_fleet_mint", "capture_env", "drop_lora_lane", "cell_lane", @@ -2063,6 +2144,7 @@ def mint_artifact( "execution_contract", "execution_contract_digest", "family_from_ref", + "finish_fleet_mint", "parse_cell_ref", "find_artifact", "flavor_label", diff --git a/src/gen_worker/executor.py b/src/gen_worker/executor.py index 677ffeed..be6a28b7 100644 --- a/src/gen_worker/executor.py +++ b/src/gen_worker/executor.py @@ -1481,11 +1481,25 @@ def _selection_for( 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). + + ``mint`` is either a finalized ``fleet_cells.SelfMint`` (artifact + already packed, digest known) or a ``fleet_cells.PendingSelfMint`` + (gw#587 CORRECT FIX: armed for capture, not yet proven or packed — its + ``target`` path exists on disk only once the warmup proof finalizes + it, and its ``snapshot_digest`` is empty until then). Either way the + ``ref`` is known immediately (computed from static axes, never the + traced FX graph bytes), which is what the hub's self-attested dispatch + fence needs at advertise time; the proof loop replaces this selection + with the fully finalized one once it packs the proven capture. """ if mint is not None: + path = getattr(mint, "artifact", None) + if path is None: + path = mint.target return _CompileArtifactSelection( - path=Path(mint.artifact), ref=str(mint.ref), - snapshot_digest=str(mint.snapshot_digest), self_mint=True) + path=Path(path), ref=str(mint.ref), + snapshot_digest=str(getattr(mint, "snapshot_digest", "") or ""), + self_mint=True) return delivered @@ -1596,6 +1610,11 @@ class _InjectionResult: active_compile_artifacts: Dict[int, _CompileArtifactSelection] = dc_field( default_factory=dict) trt_execution_before: Dict[int, int] = dc_field(default_factory=dict) + # gw#587 CORRECT FIX: id(pipeline) -> fleet_cells.PendingSelfMint for + # objects armed from a fresh self-mint capture, not yet proven or + # packed. The warmup-proof loop finalizes (packs + publishes) exactly + # the proven entries and abandons the rest — never before the proof. + pending_self_mints: Dict[int, Any] = dc_field(default_factory=dict) def add_compile_object( self, pipeline: Any, slots: typing.Iterable[str], @@ -2329,6 +2348,16 @@ def _clear_recovered_compile_failures(self, rec: _ClassRecord) -> None: self.unavailable.pop(name, None) self._compile_failure_owners.pop(name, None) + def _abandon_pending_mint(self, inj: "_InjectionResult", pipe: Any) -> None: + """Discard an unfinalized self-mint capture for ``pipe`` (gw#587 + CORRECT FIX): a disproven/unexercised candidate's capture must never + be packed or published — only a passed proof produces the mint.""" + pending = inj.pending_self_mints.pop(id(pipe), None) + if pending is not None: + from . import fleet_cells as fleet_cells_mod + + fleet_cells_mod.abandon_self_mint(pending) + def _bind_compile_guard( self, rec: _ClassRecord, target: _CompileTargetRecord, ) -> bool: @@ -3128,6 +3157,7 @@ async def _run_synthesized_warmup( snapshots: Optional[Dict[str, pb.Snapshot]], *, proof_objects: typing.Iterable[Any] = (), + cold_proof_ids: typing.Container[int] = (), ) -> _WarmupEvidence: """Run the declared per-handler warmup contract pre-READY. @@ -3135,6 +3165,12 @@ async def _run_synthesized_warmup( objects served each handler. A sibling handler is never certified by another handler's cache hit merely because both share config or an instance. Output remains local and discarded. + + ``cold_proof_ids`` (gw#587 CORRECT FIX): object ids armed from a + FRESH self-mint capture — for them a successful compiled call is the + proof (there is nothing pre-existing on disk to HIT against; the + capture this very call populates becomes the cell). Delivered cells + keep requiring a real cache hit. """ from . import compile_cache, trt_engine @@ -3182,7 +3218,10 @@ async def _run_synthesized_warmup( calls_before, trt_before = before[id(obj)] inductor_proven = ( compile_cache.execution_count(obj) > calls_before - and compile_cache.cache_hit_count(obj) > 0 + and ( + compile_cache.cache_hit_count(obj) > 0 + or id(obj) in cold_proof_ids + ) ) trt_proven = trt_engine.execution_count(obj) > trt_before if inductor_proven or trt_proven: @@ -3361,13 +3400,19 @@ async def _setup_locked( if not compile_cache.has_compile_target(pipe, spec.compile): continue inj.add_compile_object(pipe, self_loaded_slots) - selection = _selection_for( - compile_selection, scope_mints.get(id(pipe))) + mint = scope_mints.get(id(pipe)) + selection = _selection_for(compile_selection, mint) 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)) + # gw#587 CORRECT FIX: a PendingSelfMint is not proven + # or packed yet — the warmup proof below finalizes it + # (pack + publish) only after confirming a real + # compiled call, never before. + if hasattr(mint, "capture_dir"): + inj.pending_self_mints[id(pipe)] = mint # 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 @@ -3414,6 +3459,7 @@ async def run_warmup() -> Tuple[int, Dict[int, set[str]]]: candidate.pipeline for candidate in inj.compile_objects if id(candidate.pipeline) in inj.active_compile_artifacts ), + cold_proof_ids=frozenset(inj.pending_self_mints), ) return evidence.count, evidence.functions_by_object @@ -3452,8 +3498,41 @@ async def run_warmup() -> Tuple[int, Dict[int, set[str]]]: pipe_misses = compile_cache.cache_miss_count(pipe) - before[1] hits += max(0, pipe_hits) misses += max(0, pipe_misses) + pending_mint = inj.pending_self_mints.get(id(pipe)) if not warmed or calls <= 0: unexercised.append(candidate) + elif pending_mint is not None: + # gw#587 CORRECT FIX: a fresh self-mint capture has + # nothing pre-existing on disk to HIT against — its + # own real, successful compiled call (calls>0, no + # guard failure) IS the entire proof; requiring a + # disk hit here would fail every honest self-mint by + # construction. Pack the capture dir that call just + # populated and advertise the real digest — this is + # "prove-produces-the-mint": the published artifact + # is byte-derived from exactly this execution, never + # a second, separately-shaped compile. A pack + # failure never un-serves the request (the compiled + # callable is already live); it only forfeits + # advertising/publishing this boot's capture. + from . import fleet_cells as fleet_cells_mod + + finalized = fleet_cells_mod.finalize_self_mint( + pipe, pending_mint) + inj.pending_self_mints.pop(id(pipe), None) + if finalized is not None: + proven += 1 + if callable(warmup): + function_proofs[id(pipe)] = {spec.name} + inj.active_compile_artifacts[id(pipe)] = ( + _CompileArtifactSelection( + path=Path(finalized.artifact), + ref=str(finalized.ref), + snapshot_digest=str( + finalized.snapshot_digest), + self_mint=True)) + else: + disproven.append(candidate) elif pipe_hits <= 0: disproven.append(candidate) else: @@ -3479,6 +3558,7 @@ 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) + self._abandon_pending_mint(inj, pipe) detail = ( f"{len(unproven)} attached compile object(s) did not " "serve their own warmup graph " @@ -3495,6 +3575,43 @@ async def run_warmup() -> Tuple[int, Dict[int, set[str]]]: mandatory = pipeline_weight_lane(pipe).startswith( _MANDATORY_LANES) if mandatory: + pending_mint = inj.pending_self_mints.pop( + id(pipe), None) + if pending_mint is not None: + finalized = pending_mint._state.get("minted") + if finalized is not None: + # A proven sibling finalized the SHARED + # capture (same key, one family cell) — + # advertise the finalized identity, not the + # arm-time placeholder. + inj.active_compile_artifacts[id(pipe)] = ( + _CompileArtifactSelection( + path=Path(finalized.artifact), + ref=str(finalized.ref), + snapshot_digest=str( + finalized.snapshot_digest), + self_mint=True)) + else: + # Its capture was never exercised and no + # sibling finalized it: there is no proven + # artifact to advertise. Drop the target + # loudly rather than advertise bytes + # nothing proved (the gw#586 shape). + from . import fleet_cells as fleet_cells_mod + + fleet_cells_mod.abandon_self_mint( + pending_mint) + logger.warning( + "compile object (slots=%s) self-mint " + "capture unexercised (calls=0) with no " + "finalized sibling; dropping its " + "compile target", + sorted(candidate.slots)) + function_proofs[id(pipe)] = set() + compile_cache.unwrap(pipe) + inj.active_compile_artifacts.pop( + id(pipe), None) + continue # Eager is not a lane for it and a proven sibling # vouches for the cell: stays armed unproven. Its # own graphs, absent from the cell by design, fail @@ -3514,6 +3631,7 @@ 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) + self._abandon_pending_mint(inj, pipe) if ( proven and compile_selection is not None @@ -4593,6 +4711,12 @@ async def _injection_kwargs( if trt_engine.is_engine_ref(selection.ref): result.trt_execution_before[id(pipe)] = ( trt_engine.execution_count(pipe)) + # gw#587 CORRECT FIX: a PendingSelfMint is not + # proven or packed yet — the warmup proof + # finalizes it (pack + publish) only after + # confirming a real compiled call, never before. + if hasattr(pipe_mint, "capture_dir"): + result.pending_self_mints[id(pipe)] = pipe_mint delta = max(0, self._vram_allocated() - before) if slot_share: lane_obj, lane_bytes = self._register_lane( diff --git a/src/gen_worker/fleet_cells.py b/src/gen_worker/fleet_cells.py index 976d50f2..9716bf1c 100644 --- a/src/gen_worker/fleet_cells.py +++ b/src/gen_worker/fleet_cells.py @@ -11,16 +11,27 @@ 1. HIT (hub attached the cell for this runtime's self-computed key): arm through the delivered-cell path — today's behavior, unchanged. - 2. MISS: compile locally over the declared shape table (the mint brain - shared with cozy-local, ``compile_cache.mint_artifact``), adopt the - just-minted artifact, and serve COMPILED — never eager, never a - fail-closed wait. The cold-compile tax is paid once fleet-wide - instead of silently on every boot. - 3. PUBLISH (best-effort, in the background): ship the packed cell - through the hub's attested publish gate (th#910) so the next worker - on this key is store-served. Publish failure NEVER affects serving — - the local adoption already happened; a refusal (untrusted tier, - attestation, quota) is the hub's call and is fully recorded hub-side. + 2. MISS — prove-produces-the-mint (gw#587 CORRECT FIX): arm the + pipeline COLD into a fresh capture dir with NO synthetic warm call + (``compile_cache.begin_fleet_mint``). The executor's real warmup + proof — the endpoint's own serving code, the exact call shapes + production requests make — performs the only compile the mint ever + sees. The old design ran the producer-style ``_warm_call`` loop + first and proved afterwards; its synthetic call traced different FX + graphs than a conditioned/two-stage endpoint warmup (the gw#586 + defect class resurfacing inside self-mint), so the proof correctly + refused its own artifact. Now the artifact is byte-derived from the + same execution the proof observed — there is no second code path + that re-creates serving's execution to drift from. + 3. FINALIZE + PUBLISH, only after the proof PASSES + (``finalize_self_mint``): pack the proven capture, advertise its + real digest, then ship it through the hub's attested publish gate + (th#910) in the background so the next worker on this key is + store-served. A failed proof abandons the capture — nothing + unproven is ever packed or published (this also closes the old + publish-before-proof window). Publish failure NEVER affects serving; + a refusal (untrusted tier, attestation, quota) is the hub's call and + is fully recorded hub-side. The publish transport reuses the existing repo-commit machinery (``convert.hub.HubClient``) with a capability token minted by @@ -39,14 +50,14 @@ from __future__ import annotations +import dataclasses import logging import shutil import tempfile import threading -import time from dataclasses import dataclass from pathlib import Path -from typing import Any, Callable, Optional +from typing import Any, Callable, Dict, Optional from . import compile_cache as cc from .models import provision @@ -56,13 +67,17 @@ @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. + """Identity of one successfully adopted, FINALIZED self-minted cell. + + Produced only by :func:`finalize_self_mint`, after the executor's + warmup proof confirms the real serving call exercised the compiled + targets it identifies. The serving-bootstrap half of gw#587/th#910: + the minting worker ADVERTISES this identity 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 @@ -72,18 +87,55 @@ class SelfMint: artifact: Path +@dataclass(frozen=True) +class PendingSelfMint: + """A self-mint ARMED for capture, not yet proven or packed (gw#587 + CORRECT FIX). + + ``enable_compiled`` returns this on a miss instead of an already- + packed :class:`SelfMint`: the pipeline is armed cold, pointed at + ``capture_dir``, with NO synthetic warm call run against it. Only the + executor's real warmup proof — the endpoint's own serving code — + performs the compile this mint will ever see. ``ref`` is computable + immediately (STATIC axes: sku/torch/image/weight-lane/shapes/graph + structure — never the traced FX graph bytes), so the worker can + advertise its claimed key ref at arm time; ``finalize_self_mint`` + packs the artifact and computes the real digest only after the proof + passes, and publishes only from that proven capture. + + One instance may be SHARED by several pipelines of one record whose + axes compute the same key (the qwen edit shape: two lanes, one family + cell) — they cold-compile into the one capture during the one warmup + window, and the packed cell is their union. ``_state`` memoizes the + finalize outcome so sibling candidates converge on one pack/publish. + """ + + family: str + cell_key: str + ref: str + cfg: Any + target: Path + capture_dir: Path + mint_root: Path + publisher: Optional["CellPublisher"] + cache_dir: Optional[Path] = None + _state: Dict[str, Any] = dataclasses.field(default_factory=dict) + + @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. + cell) — either a :class:`PendingSelfMint` (fresh arm, not yet proven) + or, from callers that already hold a finalized identity, a + :class:`SelfMint` — letting the executor synthesize the artifact + selection it records/advertises. """ armed: bool - self_mint: Optional[SelfMint] = None + self_mint: Optional[Any] = None def __bool__(self) -> bool: return self.armed @@ -91,6 +143,12 @@ def __bool__(self) -> bool: _INTENT_TIMEOUT_S = 30 _COMPLETE_TIMEOUT_S = 30 +# Live self-mint captures by cell key. The inductor capture dir is process- +# global (one TORCHINDUCTOR_CACHE_DIR), so at most one key's capture may be +# live at a time; same-key sibling pipes join the existing capture. +_PENDING_LOCK = threading.Lock() +_PENDING: Dict[str, "PendingSelfMint"] = {} + class CellPublishRefused(Exception): """Typed hub refusal (attestation / trust tier / quota). Terminal for @@ -290,69 +348,170 @@ def enable_compiled( if bucket: cc.apply_lora_lane(pipe, bucket) - mint_root = Path(tempfile.mkdtemp(prefix="selfmint-")) - label = cc.flavor_label( - cc.runtime_key()["sku"], cc.runtime_key()["torch"], - pipeline_weight_lane(pipe)) - target = mint_root / f"{label}.tar.gz" - started = time.monotonic() - try: - meta = cc.mint_artifact(pipe, cfg, family, target, mint_root / "capture") - except Exception as exc: # noqa: BLE001 — mint failure => miss policy - logger.warning("fleet-cells: self-mint failed (%s)", exc) - cc.unwrap(pipe) - if bucket: - cc.drop_lora_lane(pipe) - return _fail_closed(pipe, f"self-mint failed: {exc}") - logger.info( - "fleet-cells: self-minted %s cell in %.0fs (%.1f MB) — adopting + publishing", - family, time.monotonic() - started, target.stat().st_size / 1e6) + # ``cell_key`` is computable from STATIC axes (sku/torch/image/weight + # lane/declared shapes+targets/module structure) — never the traced FX + # graph bytes — so the ref the hub's self-attested dispatch fence needs + # is known BEFORE any compile has happened. + from . import cell_key as cell_key_mod - # Adopt the just-minted cell through the delivered-cell path (drops the - # unguarded mint wrappers; re-traces hit the captured FX cache). This - # also proves the artifact round-trips before anyone else can pull it. - cc.unwrap(pipe) try: - armed = cc.enable(pipe, cfg, cache_dir, artifact=target) - except cc.CellSelectionBugError: + key = cell_key_mod.compute( + family, pipeline_weight_lane(pipe), bucket, + regional=bool(getattr(cfg, "regional", False)), + ).digest + except Exception as exc: # noqa: BLE001 — key axes must be computable + logger.warning("fleet-cells: self-mint key computation failed (%s)", exc) if bucket: cc.drop_lora_lane(pipe) - raise - except cc.CompiledLaneUnavailableError as exc: - logger.warning("fleet-cells: minted cell failed re-adoption (%s)", exc) + return _fail_closed(pipe, f"self-mint key computation failed: {exc}") + + # gw#587 CORRECT FIX (the defect this replaces: the old design minted + # via a separate producer-style ``mint_artifact``/``_warm_call`` BEFORE + # the real serving warmup ran — a synthetic single-stage call that can + # trace DIFFERENT FX graphs than a conditioned/two-stage endpoint's own + # warmup (the gw#586 defect class, live-found resurfacing inside self- + # mint). Arm cold instead: the caller's real warmup — run by the + # executor immediately after this returns — is the ONLY compile this + # mint will ever see, so the eventual capture is byte-derived from + # exactly the execution the proof observes. Nothing is packed or + # published here; ``finalize_self_mint`` does that, and only after the + # proof passes. + # + # The inductor capture dir is process-global (one TORCHINDUCTOR_CACHE_DIR) + # so at most ONE capture key can be live at a time: sibling pipes of the + # same record computing the SAME key share the one capture (their union + # is the family cell — the qwen edit shape); a DIFFERENT key while a + # capture is pending declines loudly into the ordinary miss policy (a + # second dir would corrupt the first capture's byte-derivation). + with _PENDING_LOCK: + existing = _PENDING.get(key) + conflict = next((k for k in _PENDING if k != key), None) + if conflict is not None and existing is None: + logger.warning( + "fleet-cells: self-mint declined for %s key=%s — capture already " + "pending for key=%s (one inductor capture dir per process)", + family, key, conflict) if bucket: cc.drop_lora_lane(pipe) - raise - if not armed: + return _fail_closed( + pipe, f"another self-mint capture is pending (key {conflict})") + + if existing is not None: + mint_root, capture_dir = existing.mint_root, existing.capture_dir + target = existing.target + else: + mint_root = Path(tempfile.mkdtemp(prefix="selfmint-")) + capture_dir = mint_root / "capture" + label = cc.flavor_label( + cc.runtime_key()["sku"], cc.runtime_key()["torch"], + pipeline_weight_lane(pipe)) + target = mint_root / f"{label}.tar.gz" + + try: + cc.begin_fleet_mint(pipe, cfg, capture_dir) + except Exception as exc: # noqa: BLE001 — arm failure => miss policy + logger.warning("fleet-cells: self-mint arm failed (%s)", exc) + if existing is None: + shutil.rmtree(mint_root, ignore_errors=True) if bucket: cc.drop_lora_lane(pipe) - return _fail_closed(pipe, "minted cell failed re-adoption") + return _fail_closed(pipe, f"self-mint arm failed: {exc}") - # 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 + if existing is not None: + logger.info( + "fleet-cells: joined pending self-mint capture for %s (key=%s)", + family, key) + return ArmOutcome(armed=True, self_mint=existing) + + pending = PendingSelfMint( + family=family, cell_key=key, ref=f"{cc.system_repo(family)}#{key}", + cfg=cfg, target=target, capture_dir=capture_dir, mint_root=mint_root, + publisher=publisher, cache_dir=cache_dir, + ) + with _PENDING_LOCK: + _PENDING[key] = pending + logger.info( + "fleet-cells: armed self-mint capture for %s (key=%s) — the real " + "warmup proof performs the only compile this mint will see", + family, key) + return ArmOutcome(armed=True, self_mint=pending) + + +def finalize_self_mint(pipe: Any, pending: "PendingSelfMint") -> Optional[SelfMint]: + """Pack + publish a self-mint AFTER the executor's warmup proof passes. + + Called from the executor's warmup-proof loop, per proven candidate — + never before the proof confirms a real, successful compiled call on + ``pipe``. Memoized on the pending object: when several sibling pipes + share one capture (same key), the first proven candidate packs and + publishes; later siblings receive the same finalized identity without + re-packing (the pack runs after the WHOLE warmup, so it already holds + every sibling's graphs). + + Packing failure never un-serves the request (``pipe``'s compiled + callables are already live in-process); it only means this boot cannot + advertise/publish a cell, so the caller must treat a ``None`` return + the same as a disproven candidate (unwrap, and fail closed for + mandatory lanes — never advertise or publish an artifact nothing + proved). + """ + state = pending._state + if "minted" in state: + return state["minted"] # sibling already finalized (or failed: None) - key = str(meta.get("cell_key") or "").strip() - if not key: - from . import cell_key + try: + meta = cc.finish_fleet_mint( + pipe, pending.cfg, pending.family, pending.target, + pending.capture_dir) + except Exception as exc: # noqa: BLE001 — pack failure => caller disproves + logger.warning( + "fleet-cells: self-mint pack failed after a passed proof (%s) — " + "the compiled callables stay live for this process, but this " + "boot cannot advertise or publish a cell", exc) + state["minted"] = None + _unregister(pending) + shutil.rmtree(pending.mint_root, ignore_errors=True) + return None - key = cell_key.from_artifact_metadata(meta).digest + from .convert.hub import blake3_file + + key = str(meta.get("cell_key") or "").strip() or pending.cell_key minted = SelfMint( - family=family, - cell_key=key, - ref=f"{cc.system_repo(family)}#{key}", - snapshot_digest="blake3:" + blake3_file(target), - artifact=target, + family=pending.family, cell_key=key, + ref=f"{cc.system_repo(pending.family)}#{key}", + snapshot_digest="blake3:" + blake3_file(pending.target), + artifact=pending.target, ) + state["minted"] = minted + _unregister(pending) + logger.info( + "fleet-cells: self-mint proof passed for %s (key=%s, %.1f MB) — " + "serving compiled, publishing", + pending.family, key, pending.target.stat().st_size / 1e6) + + # Hygiene: fold the proven capture into the live compile-cache root and + # re-point inductor there (the same end state the delivered-cell adoption + # path leaves), so later boots/adoptions in this process are not aimed at + # the soon-to-be-deleted temp capture dir. Best-effort — the in-process + # compiled callables never depend on it. + try: + live_root = ( + Path(pending.cache_dir) if pending.cache_dir + else Path.home() / ".cache" / "gen-worker") / "compile-cache" + with cc._SEED_ARM_LOCK: + cc._merge_staged_cache(pending.capture_dir, live_root) + cc.seed_env(live_root) + except Exception: + logger.debug( + "fleet-cells: live-cache fold of the proven capture failed", + exc_info=True) # 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). + publisher = pending.publisher if publisher is not None and publisher.enabled(): - _publish_async(publisher, family, target, meta) + _publish_async(publisher, pending.family, pending.target, meta) else: # Runtime assertion (gw#587): every fleet cell miss must produce a # publish attempt. A fleet worker minting with no usable sink is a @@ -361,8 +520,28 @@ def enable_compiled( # legitimately has no publisher, but it never enters this module.) logger.warning( "fleet-cells: SELF_MINT_WITHOUT_PUBLISH_SINK family=%s — cell " - "stays local to this pod; the fleet store gains nothing", family) - return ArmOutcome(armed=True, self_mint=minted) + "stays local to this pod; the fleet store gains nothing", + pending.family) + shutil.rmtree(pending.mint_root, ignore_errors=True) + return minted + + +def abandon_self_mint(pending: "PendingSelfMint") -> None: + """Discard a self-mint capture the proof did not certify (disproven or + genuinely unexercised with no proven sibling). Never packed, never + published — only the temp capture dir is cleaned up. A no-op when a + proven sibling already finalized the shared capture (the artifact and + its publish must survive).""" + if pending._state.get("minted") is not None: + return + _unregister(pending) + shutil.rmtree(pending.mint_root, ignore_errors=True) + + +def _unregister(pending: "PendingSelfMint") -> None: + with _PENDING_LOCK: + if _PENDING.get(pending.cell_key) is pending: + del _PENDING[pending.cell_key] def _fail_closed(pipe: Any, reason: str) -> ArmOutcome: @@ -393,6 +572,9 @@ def _cuda_ready() -> bool: "ArmOutcome", "CellPublishRefused", "CellPublisher", + "PendingSelfMint", "SelfMint", + "abandon_self_mint", "enable_compiled", + "finalize_self_mint", ] diff --git a/tests/test_executor_adopt.py b/tests/test_executor_adopt.py index e05d4022..ef0270c4 100644 --- a/tests/test_executor_adopt.py +++ b/tests/test_executor_adopt.py @@ -1288,6 +1288,235 @@ def _minting_enable(pipeline, *_args): assert ex.unavailable[spec.name][0] == "compile_cell_failed" +def _pending_mint_rig(tmp_path, monkeypatch, *, pipe, publisher): + """Wire the REAL fleet_cells miss path (prove-produces-the-mint) into an + Executor boot: delivered arm refuses (mandatory miss), the cold-capture + arm opens a real capture dir, and the executor's own warmup proof is the + only thing that can finalize/publish it.""" + from gen_worker import cell_key as cell_key_mod + from gen_worker import fleet_cells + + with fleet_cells._PENDING_LOCK: + fleet_cells._PENDING.clear() + + def _mandatory_miss(*a, **k): + raise cc.CompiledLaneUnavailableError("no delivered cell") + + monkeypatch.setattr(provision, "enable_compiled", _mandatory_miss) + monkeypatch.setattr(fleet_cells, "_cuda_ready", lambda: True) + monkeypatch.setattr(cc, "toolchain_present", lambda: True) + + class _Key: + digest = "ck1-" + "9" * 56 + + monkeypatch.setattr(cell_key_mod, "compute", lambda *a, **k: _Key()) + captured: dict = {} + + def _begin(p, cfg, capture): + capture.mkdir(parents=True, exist_ok=True) + captured["dir"] = capture + _mark_fake_guard(p) + + monkeypatch.setattr(cc, "begin_fleet_mint", _begin) + return captured, _Key.digest + + +def test_pending_self_mint_boot_packs_and_publishes_only_the_proven_capture( + tmp_path, monkeypatch, caplog, +): + """gw#587 CORRECT FIX direction (a), over the real Executor rig: a + mandatory-lane miss arms a COLD capture; the endpoint's own warmup — + the same execution the proof observes — produces the inductor output; + only after the proof certifies a successful compiled call does the boot + pack THAT capture, advertise its real digest, and publish those exact + bytes. Reverting finalize back into the arm path (mint-then-prove) + turns this red: the publish would happen before the warmup.""" + import gen_worker.executor as executor_mod + from gen_worker import fleet_cells + + model_dir = tmp_path / "model" + model_dir.mkdir() + events: list = [] + published: dict = {} + + class _Pub(fleet_cells.CellPublisher): + def publish(self, family, artifact, meta): + events.append("publish") + published["bytes"] = Path(artifact).read_bytes() + published["meta"] = dict(meta) + published["family"] = family + return "cp-1" + + pub = _Pub(base_url="http://hub", worker_jwt=lambda: "jwt", + image_digest="sha256:img") + pipe = _LoadablePipe() + setattr(pipe, "_cozy_weight_lane", "w8a8") + captured, mint_key = _pending_mint_rig( + tmp_path, monkeypatch, pipe=pipe, publisher=pub) + + class _MintingEndpoint(_ColdEndpoint): + def warmup(self) -> None: + type(self).warmups += 1 + events.append("warmup") + # THE real serving execution: cold-compiles the serving graphs + # into the live capture dir (successful compiled call, all + # misses — the honest cold-mint signature). + cap = captured["dir"] + (cap / "inductor" / "g").mkdir(parents=True, exist_ok=True) + (cap / "inductor" / "g" / "serving_graph.py").write_text("real") + (cap / "triton").mkdir(exist_ok=True) + _record_fake_warm(self.pipeline, hits=0, misses=8) + + spec = EndpointSpec( + name="cold-generate", method=_MintingEndpoint.run, kind="inference", + payload_type=_In, output_mode="single", cls=_MintingEndpoint, + attr_name="run", + models={"pipeline": Hub("acme/klein-finetune", flavor="fp8-w8a8")}, + compile=Compile(shapes=((768, 768),), family=FAMILY), + ) + model_ref = wire_ref(spec.models["pipeline"]) + + async def _send(_msg): + return None + + ex = Executor([spec], _send) + ex.store._cache_dir = tmp_path / "cas" + + async def _download(ref, **kwargs): + return model_dir + + monkeypatch.setattr(executor_mod, "ensure_local", _download) + monkeypatch.setattr( + provision, + "load_slot", + lambda *args, **kwargs: provision.SlotLoad(obj=pipe, is_pipeline=True), + ) + monkeypatch.setattr( + ex, "_enable_compiled", + lambda p, cfg, artifact: fleet_cells.enable_compiled( + p, cfg, ex.store._cache_dir, artifact, publisher=pub)) + _MintingEndpoint.setups = _MintingEndpoint.warmups = 0 + + with caplog.at_level("ERROR", logger="gen_worker.executor"): + instance = asyncio.run(ex.ensure_setup( + spec, {model_ref: pb.Snapshot(digest=MODEL_DIGEST)})) + assert isinstance(instance, _MintingEndpoint) + assert _MintingEndpoint.warmups == 1 + + # publish is async; join the publisher thread via the recorded event + for _ in range(100): + if "publish" in events: + break + import time as _time + + _time.sleep(0.05) + assert events.index("warmup") < events.index("publish"), ( + "prove-produces-the-mint: publish must come from the proven capture, " + "never precede the warmup proof") + + (target,) = ex.compile_targets() + assert target.active_compile_ref == f"_system/family-{FAMILY}#{mint_key}" + digest = target.active_compile_snapshot_digest + assert digest.startswith("blake3:") + # The advertised digest is the digest of exactly the published bytes, + # and those bytes contain the graphs the WARMUP (not any producer warm + # loop) compiled. + from gen_worker.convert.hub import blake3_file + + copy = tmp_path / "published-copy.tar.gz" + copy.write_bytes(published["bytes"]) + assert digest == "blake3:" + blake3_file(copy) + import io + import tarfile + + with tarfile.open(fileobj=io.BytesIO(published["bytes"]), mode="r:*") as tar: + names = tar.getnames() + assert any("serving_graph.py" in n for n in names) + assert published["meta"].get("source_ref") == "self-mint" + # A minting boot legitimately compiles: no store-served alarm. + assert not any( + "STORE_SERVED_BOOT_COMPILED" in r.message for r in caplog.records) + + +def test_pending_self_mint_unproven_fails_closed_and_never_publishes( + tmp_path, monkeypatch, +): + """gw#587 CORRECT FIX direction (b), revert-turns-red: a mandatory-lane + self-mint capture whose warmup never certifies a successful compiled + call (the executor ran the warmup, but the compiled targets did not + serve it — the gw#586 silent-eager shape) must fail the boot closed, + advertise nothing, publish NOTHING, and abandon the capture. If packing + or publishing ever moves ahead of the proof again, the publish below + fires and this test goes red.""" + import gen_worker.executor as executor_mod + from gen_worker import fleet_cells + + model_dir = tmp_path / "model" + model_dir.mkdir() + + class _Pub(fleet_cells.CellPublisher): + def publish(self, family, artifact, meta): + pytest.fail("an unproven self-mint must NEVER publish") + + pub = _Pub(base_url="http://hub", worker_jwt=lambda: "jwt", + image_digest="sha256:img") + pipe = _LoadablePipe() + setattr(pipe, "_cozy_weight_lane", "w8a8") + captured, _mint_key = _pending_mint_rig( + tmp_path, monkeypatch, pipe=pipe, publisher=pub) + + class _UnprovenEndpoint(_ColdEndpoint): + def warmup(self) -> None: + type(self).warmups += 1 + # Warmup runs, but no successful compiled call is ever recorded + # on the pipe (calls=0): the capture certifies nothing. + + spec = EndpointSpec( + name="cold-generate", method=_UnprovenEndpoint.run, kind="inference", + payload_type=_In, output_mode="single", cls=_UnprovenEndpoint, + attr_name="run", + models={"pipeline": Hub("acme/klein-finetune", flavor="fp8-w8a8")}, + compile=Compile(shapes=((768, 768),), family=FAMILY), + ) + model_ref = wire_ref(spec.models["pipeline"]) + + async def _send(_msg): + return None + + ex = Executor([spec], _send) + ex.store._cache_dir = tmp_path / "cas" + + async def _download(ref, **kwargs): + return model_dir + + monkeypatch.setattr(executor_mod, "ensure_local", _download) + monkeypatch.setattr( + provision, + "load_slot", + lambda *args, **kwargs: provision.SlotLoad(obj=pipe, is_pipeline=True), + ) + monkeypatch.setattr( + ex, "_enable_compiled", + lambda p, cfg, artifact: fleet_cells.enable_compiled( + p, cfg, ex.store._cache_dir, artifact, publisher=pub)) + _UnprovenEndpoint.setups = _UnprovenEndpoint.warmups = 0 + + with pytest.raises( + cc.CompiledLaneUnavailableError, + match="did not serve their own warmup graph", + ): + asyncio.run(ex.ensure_setup( + spec, {model_ref: pb.Snapshot(digest=MODEL_DIGEST)})) + assert _UnprovenEndpoint.warmups == 1, "the proof must have actually run" + assert ex.compile_targets() == [], "an unproven self-mint must not advertise" + assert ex.unavailable[spec.name][0] == "compile_cell_failed" + # The capture was abandoned, never packed. + mint_root = captured["dir"].parent + assert not mint_root.exists(), "an uncertified capture must be discarded" + with fleet_cells._PENDING_LOCK: + assert fleet_cells._PENDING == {} + + def test_boot_warmup_proves_each_compile_object_independently( tmp_path, monkeypatch, ): diff --git a/tests/test_fleet_cells.py b/tests/test_fleet_cells.py index 1376653a..d6304009 100644 --- a/tests/test_fleet_cells.py +++ b/tests/test_fleet_cells.py @@ -2,9 +2,12 @@ The torch.compile capture itself needs a GPU + toolchain and is proven live (the gw#587 live proof); these tests pin the POLICY around it: delivered -cell first, self-mint on miss, publish best-effort and never load-bearing -for serving, the cell_selection_bug receipt invariant untouched, and the -typed quantized refusal only at genuine mint impossibilities. +cell first, PROVE-PRODUCES-THE-MINT on miss (arm cold for capture; only +the executor's passed warmup proof packs + publishes — the gw#586-class +fix: no synthetic producer warm loop exists in the fleet path anymore), +publish best-effort and never load-bearing for serving, the +cell_selection_bug receipt invariant untouched, and the typed quantized +refusal only at genuine mint impossibilities. """ from __future__ import annotations @@ -15,6 +18,7 @@ import pytest +from gen_worker import cell_key from gen_worker import compile_cache as cc from gen_worker import fleet_cells as fc from gen_worker.models import provision @@ -34,6 +38,40 @@ class _Pipe: _cozy_low_vram_mode = "off" +FAKE_KEY = "ck1-" + "a" * 56 + + +@pytest.fixture(autouse=True) +def _clear_pending(): + with fc._PENDING_LOCK: + fc._PENDING.clear() + yield + with fc._PENDING_LOCK: + fc._PENDING.clear() + + +def _mintable(monkeypatch, *, key=FAKE_KEY): + """Route a MISS into the cold-capture arm with no CUDA/toolchain.""" + monkeypatch.setattr(provision, "enable_compiled", lambda *a, **k: False) + monkeypatch.setattr(fc, "_cuda_ready", lambda: True) + monkeypatch.setattr(cc, "toolchain_present", lambda: True) + + class _Key: + digest = key + + monkeypatch.setattr(cell_key, "compute", lambda *a, **k: _Key()) + + def _begin(pipe, cfg, capture): + # the real begin_fleet_mint latches env + arms cold; the capture + # content itself arrives during the (real, GPU) warmup — simulate + # the layout the proof window would produce. + (capture / "inductor" / "g").mkdir(parents=True, exist_ok=True) + (capture / "inductor" / "g" / "kernel.py").write_text("compiled") + (capture / "triton").mkdir(exist_ok=True) + + monkeypatch.setattr(cc, "begin_fleet_mint", _begin) + + def _publisher(calls): class _Pub(fc.CellPublisher): def publish(self, family, artifact, meta): @@ -52,9 +90,11 @@ def test_delivered_cell_hit_never_mints_or_publishes(monkeypatch, tmp_path): calls: list = [] monkeypatch.setattr(provision, "enable_compiled", lambda *a, **k: True) monkeypatch.setattr( - cc, "mint_artifact", - lambda *a, **k: pytest.fail("HIT must never mint")) - assert fc.enable_compiled(_Pipe(), _Cfg(), tmp_path, None, publisher=_publisher(calls)) + cc, "begin_fleet_mint", + lambda *a, **k: pytest.fail("HIT must never open a mint capture")) + outcome = fc.enable_compiled( + _Pipe(), _Cfg(), tmp_path, None, publisher=_publisher(calls)) + assert outcome and outcome.self_mint is None assert calls == [] @@ -68,58 +108,120 @@ def _raise(*a, **k): monkeypatch.setattr(provision, "enable_compiled", _raise) monkeypatch.setattr( - cc, "mint_artifact", - lambda *a, **k: pytest.fail("selection bug must never trigger a mint")) + cc, "begin_fleet_mint", + lambda *a, **k: pytest.fail("selection bug must never open a capture")) with pytest.raises(cc.CellSelectionBugError): fc.enable_compiled(_Pipe(), _Cfg(), tmp_path, None, publisher=_publisher([])) -def test_miss_mints_adopts_and_publishes(monkeypatch, tmp_path): - """The core gw#587 outcome: MISS -> local mint -> serve compiled -> - publish attempt (with the mint's own metadata).""" +def test_miss_arms_pending_capture_without_packing_or_publishing( + monkeypatch, tmp_path, +): + """gw#587 CORRECT FIX direction (b): a MISS opens a cold capture and + returns a PENDING mint — nothing is packed, nothing is published, and + no synthetic warm call runs (that separate producer-shaped execution + was the gw#586-class defect the live proof caught). Publish before the + proof reverts this test red.""" + calls: list = [] + _mintable(monkeypatch) + monkeypatch.setattr( + cc, "mint_artifact", + lambda *a, **k: pytest.fail( + "the fleet miss path must never run the producer warm loop")) + + outcome = fc.enable_compiled( + _Pipe(), _Cfg(), tmp_path, None, publisher=_publisher(calls)) + assert outcome.armed + pending = outcome.self_mint + assert isinstance(pending, fc.PendingSelfMint) + assert pending.cell_key == FAKE_KEY + assert pending.ref == f"_system/family-fam#{FAKE_KEY}" + assert not pending.target.exists(), "nothing packed before the proof" + assert calls == [], "nothing published before the proof" + + +def test_finalize_packs_the_proven_capture_and_publishes_it( + monkeypatch, tmp_path, +): + """gw#587 CORRECT FIX direction (a): after the proof passes, finalize + packs EXACTLY the capture the proof window populated and publishes + those bytes; the advertised digest is the digest of that packed + artifact.""" calls: list = [] published = threading.Event() class _Pub(fc.CellPublisher): def publish(self, family, artifact, meta): - calls.append((family, Path(artifact), dict(meta))) + calls.append((family, artifact.read_bytes(), dict(meta))) published.set() return "cp-1" - pub = _Pub(base_url="http://hub", worker_jwt=lambda: "jwt", image_digest="sha256:img") - monkeypatch.setattr(provision, "enable_compiled", lambda *a, **k: False) - monkeypatch.setattr(fc, "_cuda_ready", lambda: True) - monkeypatch.setattr(cc, "toolchain_present", lambda: True) - - def _fake_mint(pipe, cfg, family, target, capture, **kw): - target.parent.mkdir(parents=True, exist_ok=True) - target.write_bytes(b"cell-bytes") - return {"cell_key": "ck1-" + "a" * 56, "sku": "b200", "gen_worker": "0.39.0"} + pub = _Pub(base_url="http://hub", worker_jwt=lambda: "jwt", + image_digest="sha256:img") + _mintable(monkeypatch) + pipe = _Pipe() + outcome = fc.enable_compiled(pipe, _Cfg(), tmp_path, None, publisher=pub) + pending = outcome.self_mint + assert isinstance(pending, fc.PendingSelfMint) + + minted = fc.finalize_self_mint(pipe, pending) + assert minted is not None + assert minted.cell_key == FAKE_KEY + assert minted.ref == f"_system/family-fam#{FAKE_KEY}" + assert minted.snapshot_digest.startswith("blake3:") + assert len(minted.snapshot_digest) == len("blake3:") + 64 + assert published.wait(5), "a finalized mint must attempt publish" + (family, tar_bytes, meta) = calls[0] + assert family == "fam" + # The published bytes ARE the packed proven capture, and the advertised + # digest is the digest of exactly those bytes. + from gen_worker.convert.hub import blake3_file + + copy = tmp_path / "published-copy.tar.gz" + copy.write_bytes(tar_bytes) + assert minted.snapshot_digest == "blake3:" + blake3_file(copy) + import io + import tarfile + + with tarfile.open(fileobj=io.BytesIO(tar_bytes), mode="r:*") as tar: + names = tar.getnames() + assert any("kernel.py" in n for n in names), ( + "published cell must contain the capture the proof produced") + # Finalize is memoized for same-key siblings: no double pack/publish. + assert fc.finalize_self_mint(pipe, pending) is minted + assert len(calls) == 1 + + +def test_abandon_never_publishes(monkeypatch, tmp_path): + """A capture whose proof did not certify it is abandoned: nothing + packed, nothing published, temp dir removed.""" + calls: list = [] + _mintable(monkeypatch) + pipe = _Pipe() + outcome = fc.enable_compiled( + pipe, _Cfg(), tmp_path, None, publisher=_publisher(calls)) + pending = outcome.self_mint + fc.abandon_self_mint(pending) + assert calls == [] + assert not pending.mint_root.exists() + with fc._PENDING_LOCK: + assert fc._PENDING == {} - monkeypatch.setattr(cc, "mint_artifact", _fake_mint) - monkeypatch.setattr(cc, "unwrap", lambda pipe: None) - monkeypatch.setattr(cc, "enable", lambda *a, **k: True) - outcome = fc.enable_compiled(_Pipe(), _Cfg(), tmp_path, None, publisher=pub) - assert outcome.armed - assert published.wait(5), "publish attempt must be recorded on every miss" - (family, artifact, meta) = calls[0] - assert family == "fam" - assert meta["cell_key"].startswith("ck1-") - # Serving-bootstrap identity (th#910 self-attested fence): the outcome - # carries the worker's OWN key ref + a self-attested artifact digest so - # the executor can advertise the mint exactly like a delivered cell. - mint = outcome.self_mint - assert mint is not None - assert mint.ref == f"_system/family-fam#{meta['cell_key']}" - assert mint.cell_key == meta["cell_key"] - assert mint.snapshot_digest.startswith("blake3:") - assert len(mint.snapshot_digest) == len("blake3:") + 64 +def test_same_key_sibling_joins_the_pending_capture(monkeypatch, tmp_path): + """Two pipes of one record computing the same key share ONE capture + (the union family cell); a second mint_root is never created.""" + _mintable(monkeypatch) + a, b = _Pipe(), _Pipe() + first = fc.enable_compiled(a, _Cfg(), tmp_path, None).self_mint + second = fc.enable_compiled(b, _Cfg(), tmp_path, None).self_mint + assert second is first def test_publish_failure_never_affects_serving(monkeypatch, tmp_path): - """The request that triggered the miss is served from the local mint even - when the hub refuses the publish (untrusted tier / forged axis / quota).""" + """The request that triggered the miss is served from the proven local + capture even when the hub refuses the publish (untrusted tier / forged + axis / quota).""" refused = threading.Event() class _Pub(fc.CellPublisher): @@ -128,20 +230,11 @@ def publish(self, family, artifact, meta): raise fc.CellPublishRefused("cell_publish_untrusted_tier: community_tier") pub = _Pub(base_url="http://hub", worker_jwt=lambda: "jwt", image_digest="d") - monkeypatch.setattr(provision, "enable_compiled", lambda *a, **k: False) - monkeypatch.setattr(fc, "_cuda_ready", lambda: True) - monkeypatch.setattr(cc, "toolchain_present", lambda: True) - - def _fake_mint(pipe, cfg, family, target, capture, **kw): - target.parent.mkdir(parents=True, exist_ok=True) - target.write_bytes(b"cell-bytes") - return {"cell_key": "ck1-" + "b" * 56} - - monkeypatch.setattr(cc, "mint_artifact", _fake_mint) - monkeypatch.setattr(cc, "unwrap", lambda pipe: None) - monkeypatch.setattr(cc, "enable", lambda *a, **k: True) - - assert fc.enable_compiled(_Pipe(), _Cfg(), tmp_path, None, publisher=pub) + _mintable(monkeypatch) + pipe = _Pipe() + outcome = fc.enable_compiled(pipe, _Cfg(), tmp_path, None, publisher=pub) + minted = fc.finalize_self_mint(pipe, outcome.self_mint) + assert minted is not None, "hub refusal must never fail the finalize" assert refused.wait(5)