alex/club-3090
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
49d9bb4313a97d3cacd6cafdaebc5c2903fc4bfb
club-3090/scripts/tests/test-pull.sh
T

1339 lines
63 KiB
Bash
Executable File
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
#!/usr/bin/env bash
# v0.8.0 Pull-Gate P4 — the EXHAUSTIVE state-machine truth table.
#
# Per the LOCKED stop-condition (brief v6): prose iteration ended; THIS test
# is now the authoritative spec for the Pull-Gate state machine. It covers,
# with NO live network / NO GPU / NO HF-API / NO real [D]-emit:
#
# * the §4.1 9 cells of [C1] (3 confidences × 3 raw-verdicts) with the
# correct terminal + every flag interaction, asserted TOTAL + design-
# locked (terminal set EXACTLY the locked 4);
# * every stratum's abort, in order, with ordering assertions
# (stratum-4 [C2a] AFTER stratum-3 [C0]; stratum-5 after [C2a];
# stratum-6 Path-A-only);
# * --experimental-arch bypasses ONLY no-arch-row (NOT
# runtime-incompatible, NOT stratum-5 no-fit-model);
# * golden cases g0..g15 from the brief;
# * no --trust-remote-code ever in an in-scope Path-A emitted body
# (reusing the patch_attribution.service_body()==0 check pattern).
#
# Hermetic: hardware_sm injected; HF fetcher is a recorded FixtureFetcher;
# disk free is an injected fake statvfs; [D] emit goes through the REAL
# generate_compose.generate (pure, no container) for the integration
# goldens and an injected d_runner for the dry-run-refusal golden.
set -euo pipefail
ROOT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")/../.." && pwd)"
cd "$ROOT_DIR"
export PYTHONPATH="$ROOT_DIR${PYTHONPATH:+:$PYTHONPATH}"
python3 - "$ROOT_DIR" <<'PY'
from __future__ import annotations
import json
import os
import sys
from pathlib import Path
os.environ.pop("HF_TOKEN", None) # deterministic gated path
root = Path(sys.argv[1])
sys.path.insert(0, str(root))
from scripts.lib.profiles import pull as P # noqa: E402 (P4 — under test)
from scripts.lib.profiles import deriver as D # noqa: E402 (P2, frozen)
from scripts.lib.profiles import gates as G # noqa: E402 (P3, frozen)
from scripts.lib.profiles import patch_attribution as PA # noqa: E402
from scripts.lib.profiles.compat import load_profiles # noqa: E402
from scripts.lib import generate_compose as GC # noqa: E402 ([D], frozen)
failures: list[str] = []
def check(cond: bool, msg: str) -> None:
if cond:
print(f"PASS: {msg}")
else:
print(f"FAIL: {msg}", file=sys.stderr)
failures.append(msg)
profiles = load_profiles()
SM_86 = 8.6 # RTX 3090 (Ampere)
SM_90 = 9.0 # Hopper-class
CFG = f"{D._HF_RESOLVE}/{{slug}}/resolve/main/config.json"
API = f"{D._HF_API}/{{slug}}?blobs=true"
class FixtureFetcher:
"""Recorded-response fetcher (the P2/P3 pattern). Raises if a Path-A
curated run touches it (curated hits must be network-free)."""
def __init__(self, routes: dict):
self.routes = routes
self.calls: list = []
def get(self, url, headers=None, range_=None):
self.calls.append((url, range_))
if url not in self.routes:
return D.FetchResponse(status=404, body=b"")
spec = self.routes[url]
if isinstance(spec, D.FetchResponse):
return spec
if isinstance(spec, dict):
return D.FetchResponse(
status=200, body=json.dumps(spec).encode("utf-8")
)
raise AssertionError(f"bad fixture for {url}")
class NoNet:
def get(self, *a, **k):
raise AssertionError("curated Path-A run must not hit the network")
def fake_statvfs(free_gb: float):
def _sv(_p):
class S:
f_frsize = 4096
f_bavail = int(free_gb * (1024 ** 3) / 4096)
return S()
return _sv
BIG_DISK = fake_statvfs(500.0)
TINY_DISK = fake_statvfs(2.0)
def dense_cfg(arch="LlamaForCausalLM", **over):
c = {
"model_type": "llama",
"architectures": [arch],
"hidden_size": 4096,
"num_hidden_layers": 32,
"num_attention_heads": 32,
"num_key_value_heads": 8,
"max_position_embeddings": 131072,
"torch_dtype": "bfloat16",
}
c.update(over)
return c
def dense_api(weight_gb=8.0):
return {
"siblings": [
{"rfilename": "config.json", "size": 700},
{"rfilename": "tokenizer.json", "size": 1_000_000},
{"rfilename": "model.safetensors",
"size": int(weight_gb * (1024 ** 3))},
]
}
def ff_derived(slug, cfg, weight_gb=8.0):
return FixtureFetcher({
API.format(slug=slug): dense_api(weight_gb),
CFG.format(slug=slug): cfg,
})
CURATED_SLUG = "Lorbus/Qwen3.6-27B-int4-AutoRound"
CURATED_MODEL_ID = "qwen3.6-27b" # Tier-1 curated model CURATED_SLUG resolves to
CURATED_VARIANT = "autoround_int4" # variant CURATED_SLUG resolves to
# ---------------------------------------------------------------------------
# NON-MOCKED kv-calc parameterization.
#
# The pre-fix bug: `_curated_spec` priced Tier-1 curated hits through P1's
# conservative generic-dense LOWER-BOUND instead of the model's authoritative
# curated-exact family branch. The mocked truth-table never ran the real
# `_curated_spec` -> real `kv.raw_verdict` for a curated model, so a curated
# config that genuinely FITS got a false hard-block and every test stayed
# green.
#
# This helper derives the EXPECTED Pull-Gate `[B]` raw_verdict + `[C1]`
# terminal for a curated `(model_id, variant, profile-like)` straight from
# kv-calc's OWN authoritative curated-exact spec (`kv.MODEL_SPECS[...]`, the
# same specs `tools/kv-calc.py --calibration` validates at 22/22) and a LIVE
# `kv.predict()` call. Expectations are parameterized off kv-calc itself, so
# the Pull-Gate `[B]` verdict can NEVER silently diverge from kv-calc again
# (a divergence becomes a hard FAIL here, not a frozen-buggy green).
# ---------------------------------------------------------------------------
import importlib.util as _ilu # noqa: E402
_kv_path = root / "tools" / "kv-calc.py"
_kv_spec = _ilu.spec_from_file_location("kv_calc", _kv_path)
_KVC = _ilu.module_from_spec(_kv_spec)
sys.modules["kv_calc"] = _KVC # MUST precede exec_module (@dataclass)
_kv_spec.loader.exec_module(_KVC)
sys.path.insert(0, str(root / "scripts" / "lib" / "profiles"))
from compose_registry import COMPOSE_REGISTRY as _COMPOSE_REGISTRY # noqa: E402
# kv-calc raw_verdict -> Pull-Gate raw_verdict; then §4.1 exact-row terminal.
_RAWMAP = {"PASS": "fits-clean", "TIGHT": "fits-constrained",
"FAIL": "wont-fit"}
_EXACT_TERMINAL = { # confidence == "exact" rows of the §4.1 table
"fits-clean": "proceed",
"fits-constrained": "confirm→proceed",
"wont-fit": "hard-block",
}
def curated_exact_expectation(profile_like: str,
model_id: str = CURATED_MODEL_ID,
variant: str = CURATED_VARIANT):
"""Return (expected_raw_verdict, expected_exact_terminal) computed LIVE
from kv-calc's authoritative curated-exact spec — never hardcoded."""
mm = _COMPOSE_REGISTRY[profile_like]
spec = dict(_KVC.MODEL_SPECS[model_id]) # authoritative; copy
assert spec["model_family"] != "generic-dense", (
f"{model_id} curated-exact spec must NOT be generic-dense")
size_gb = profiles.models[model_id].weights[variant].get("size_gb")
if isinstance(size_gb, (int, float)) and float(size_gb) > 0:
spec["weights_total_gb"] = float(size_gb) # qwen3-next-hybrid weight field
p = _KVC.predict(
spec=spec,
kv_format=mm["kv_format"],
max_ctx=int(mm["max_ctx"]),
max_num_seqs=int(mm["max_num_seqs"] or 1),
tp=int(mm["tp"]),
mem_util=float(mm["mem_util"] or 0.95),
vram_gb=24,
)
rv = _RAWMAP[p.verdict]
return rv, _EXACT_TERMINAL[rv]
# ===========================================================================
# SECTION 1 — [C1] §4.1 TOTAL FUNCTION: all 9 cells × flag interactions.
# This is a PURE-function unit. It is the single authoritative mapping;
# the rest of the orchestrator only consumes it.
# ===========================================================================
print("\n--- [C1] §4.1 3×3 total function (pure) ---")
# design-lock: terminal set is EXACTLY the locked 4.
check(
P.LOCKED_TERMINALS == {"proceed", "confirm→proceed", "hard-block",
"override-accepted"},
f"design-lock: Terminal set == locked 4 (got {sorted(P.LOCKED_TERMINALS)})",
)
check(
{t.value for t in P.Terminal} == P.LOCKED_TERMINALS,
"design-lock: Terminal enum == LOCKED_TERMINALS (no extra states)",
)
# TOTAL: every (confidence, raw_verdict) in the domain maps to a cell.
unmapped = [
(c, v)
for c in P.C1_CONFIDENCE_DOMAIN
for v in P.C1_RAW_VERDICT_DOMAIN
if (c, v) not in P._C1_TABLE
]
check(not unmapped, f"[C1] is TOTAL over 3×3 domain (unmapped: {unmapped})")
check(
len(P._C1_TABLE) == 9,
f"[C1] table is exactly 9 cells (got {len(P._C1_TABLE)})",
)
# every cell's terminal is in the locked set
bad_term = [
k for k, c in P._C1_TABLE.items()
if c.base_terminal.value not in P.LOCKED_TERMINALS
]
check(not bad_term, f"[C1] every cell terminal ∈ locked 4 (bad: {bad_term})")
# Expected §4.1 outcomes (terminal, satisfied) per cell + flags.
# Source: v0.8.x-design.md §4.1 table (lines 62-66) + footnote line 68.
EXACT = D.Confidence.EXACT.value
DERV = D.Confidence.DERIVED.value
ELB = D.Confidence.ESTIMATED_LOWER_BOUND.value
# exact × fits-clean -> proceed, SILENT (the ONLY silent gate-pass).
o = P.c1_terminal(EXACT, "fits-clean", {})
check(o.terminal is P.Terminal.PROCEED and o.satisfied and o.needs == "",
"[C1] exact×fits-clean -> proceed (silent, no flag)")
# exact × fits-constrained -> confirm→proceed; needs --yes.
o = P.c1_terminal(EXACT, "fits-constrained", {})
check(o.terminal is P.Terminal.CONFIRM_PROCEED and not o.satisfied
and o.needs == "--yes",
"[C1] exact×fits-constrained -> confirm→proceed, needs --yes")
o = P.c1_terminal(EXACT, "fits-constrained", {"yes": True})
check(o.terminal is P.Terminal.CONFIRM_PROCEED and o.satisfied,
"[C1] exact×fits-constrained + --yes -> satisfied")
# exact × wont-fit -> hard-block (no flag clears it).
o = P.c1_terminal(EXACT, "wont-fit", {"yes": True, "force_download": True})
check(o.terminal is P.Terminal.HARD_BLOCK and not o.satisfied,
"[C1] exact×wont-fit -> hard-block (NO flag clears it)")
# derived × fits-clean -> confirm→proceed, needs --yes.
o = P.c1_terminal(DERV, "fits-clean", {})
check(o.terminal is P.Terminal.CONFIRM_PROCEED and not o.satisfied
and o.needs == "--yes",
"[C1] derived×fits-clean -> confirm→proceed, needs --yes")
o = P.c1_terminal(DERV, "fits-clean", {"yes": True})
check(o.satisfied, "[C1] derived×fits-clean + --yes -> satisfied")
# derived × fits-constrained -> confirm→proceed, needs --yes.
o = P.c1_terminal(DERV, "fits-constrained", {})
check(o.terminal is P.Terminal.CONFIRM_PROCEED and not o.satisfied,
"[C1] derived×fits-constrained -> confirm→proceed, needs --yes")
# derived × wont-fit -> advisory → --force-download → override-accepted.
o = P.c1_terminal(DERV, "wont-fit", {})
check(o.terminal is P.Terminal.OVERRIDE_ACCEPTED and not o.satisfied
and o.needs == "--force-download",
"[C1] derived×wont-fit -> override-accepted, needs --force-download")
o = P.c1_terminal(DERV, "wont-fit", {"force_download": True})
check(o.terminal is P.Terminal.OVERRIDE_ACCEPTED and o.satisfied,
"[C1] derived×wont-fit + --force-download -> override-accepted")
# --yes alone does NOT clear a wont-fit advisory.
o = P.c1_terminal(DERV, "wont-fit", {"yes": True})
check(not o.satisfied,
"[C1] derived×wont-fit + --yes (no --force-download) -> NOT satisfied")
# estimated-lower-bound × fits-clean -> confirm→proceed, needs --yes.
o = P.c1_terminal(ELB, "fits-clean", {})
check(o.terminal is P.Terminal.CONFIRM_PROCEED and not o.satisfied
and o.needs == "--yes",
"[C1] estimated-lower-bound×fits-clean -> confirm→proceed, needs --yes")
o = P.c1_terminal(ELB, "fits-clean", {"yes": True})
check(o.satisfied,
"[C1] estimated-lower-bound×fits-clean + --yes -> satisfied")
# estimated-lower-bound × fits-constrained -> confirm→proceed, needs --yes.
o = P.c1_terminal(ELB, "fits-constrained", {})
check(o.terminal is P.Terminal.CONFIRM_PROCEED and not o.satisfied,
"[C1] estimated-lower-bound×fits-constrained -> confirm→proceed")
# estimated-lower-bound × wont-fit -> advisory → --force-download.
o = P.c1_terminal(ELB, "wont-fit", {})
check(o.terminal is P.Terminal.OVERRIDE_ACCEPTED and not o.satisfied
and o.needs == "--force-download",
"[C1] estimated-lower-bound×wont-fit -> override-accepted, "
"needs --force-download")
o = P.c1_terminal(ELB, "wont-fit", {"force_download": True})
check(o.terminal is P.Terminal.OVERRIDE_ACCEPTED and o.satisfied,
"[C1] estimated-lower-bound×wont-fit + --force-download -> "
"override-accepted")
# §4.1 footnote line 68: ONLY exact×fits-clean reaches proceed without --yes.
silent = [
(c, v)
for c in P.C1_CONFIDENCE_DOMAIN
for v in P.C1_RAW_VERDICT_DOMAIN
if P.c1_terminal(c, v, {}).satisfied
and P.c1_terminal(c, v, {}).terminal is P.Terminal.PROCEED
]
check(silent == [(EXACT, "fits-clean")],
f"[C1] ONLY exact×fits-clean silent-passes (got {silent})")
# ===========================================================================
# SECTION 2 — STRATUM 1: deriver structured errors (g12 trio + stratum-1).
# Aborts BEFORE any gate.
# ===========================================================================
print("\n--- stratum-1: deriver structured errors (g12) ---")
# g12a: no *.safetensors -> unsupported-format.
s = "fixtures/gguf-only"
ff = FixtureFetcher({
API.format(slug=s): {"siblings": [{"rfilename": "model.gguf",
"size": 9}]},
CFG.format(slug=s): dense_cfg(),
})
r = P.run_pull(s, "vllm/minimal", hardware_sm=SM_86, fetcher=ff,
profiles=profiles, statvfs=BIG_DISK)
check(r.stratum is P.Stratum.DERIVER
and r.abort_reason == "unsupported-format" and not r.ok,
f"g12: no safetensors -> stratum-1 unsupported-format "
f"(got {r.stratum.name}/{r.abort_reason})")
# g12b: multiple top-level safetensors, no index -> ambiguous-weight-set.
s = "fixtures/ambiguous"
ff = FixtureFetcher({
API.format(slug=s): {"siblings": [
{"rfilename": "modelA.safetensors", "size": 9000000000},
{"rfilename": "modelB.safetensors", "size": 9000000000},
{"rfilename": "config.json", "size": 700},
]},
CFG.format(slug=s): dense_cfg(),
})
r = P.run_pull(s, "vllm/minimal", hardware_sm=SM_86, fetcher=ff,
profiles=profiles, statvfs=BIG_DISK)
check(r.stratum is P.Stratum.DERIVER
and r.abort_reason == "ambiguous-weight-set",
f"g12: multiple sets -> stratum-1 ambiguous-weight-set "
f"(got {r.stratum.name}/{r.abort_reason})")
# g12c: repo-not-found (HF model API 404).
s = "fixtures/nope"
ff = FixtureFetcher({API.format(slug=s): D.FetchResponse(404, b"")})
r = P.run_pull(s, "vllm/minimal", hardware_sm=SM_86, fetcher=ff,
profiles=profiles, statvfs=BIG_DISK)
check(r.stratum is P.Stratum.DERIVER
and r.abort_reason == "repo-not-found",
f"g12: 404 -> stratum-1 repo-not-found "
f"(got {r.stratum.name}/{r.abort_reason})")
# stratum-1 aborts BEFORE [C0] / [B] (no fabricated verdict, §1).
check(r.raw_verdict is None and r.terminal is None,
"g12: stratum-1 abort emits NO fit verdict (§1 honesty)")
# ===========================================================================
# SECTION 3 — STRATUM 2: --profile-like precondition (g0, g3b, g13).
# Both paths, BEFORE [C0].
# ===========================================================================
print("\n--- stratum-2: --profile-like precondition (g0/g3b/g13) ---")
# g13: non-vLLM --profile-like (llamacpp/default) -> unsupported-runtime-engine
# (both paths), BEFORE [C0]/[B].
r = P.run_pull(CURATED_SLUG, "llamacpp/default", path="B",
hardware_sm=SM_86, fetcher=NoNet(), profiles=profiles,
statvfs=BIG_DISK)
check(r.stratum is P.Stratum.PROFILE_LIKE
and r.abort_reason == "unsupported-runtime-engine",
f"g13: llamacpp/default -> stratum-2 unsupported-runtime-engine "
f"(got {r.stratum.name}/{r.abort_reason})")
# g0: Path-A non-emittable profile (vllm/dual-turbo, genesis_equipped) ->
# stratum-2 profile-not-emittable BEFORE [C0].
r = P.run_pull(CURATED_SLUG, "vllm/dual-turbo", path="A",
hardware_sm=SM_86, fetcher=NoNet(), profiles=profiles,
statvfs=BIG_DISK)
check(r.stratum is P.Stratum.PROFILE_LIKE
and r.abort_reason == "profile-not-emittable",
f"g0: vllm/dual-turbo (genesis) -> stratum-2 profile-not-emittable "
f"(got {r.stratum.name}/{r.abort_reason})")
# g3b: Path-A model/variant mismatch -> stratum-2 profile-mismatch.
# vllm/gemma-mtp is vLLM + emittable but model=gemma-4-31b != curated qwen.
r = P.run_pull(CURATED_SLUG, "vllm/gemma-mtp", path="A",
hardware_sm=SM_86, fetcher=NoNet(), profiles=profiles,
statvfs=BIG_DISK)
check(r.stratum is P.Stratum.PROFILE_LIKE
and r.abort_reason == "profile-mismatch",
f"g3b: model/variant mismatch -> stratum-2 profile-mismatch "
f"(got {r.stratum.name}/{r.abort_reason})")
# ===========================================================================
# SECTION 4 — STRATUM 3: [C0] engine-support / runtime / SM
# (g4, g5, g6a, g6b, g10, g14) + flag-bypass scoping.
# ===========================================================================
print("\n--- stratum-3: [C0] (g4/g5/g6a/g6b/g10/g14) ---")
# g4: Llama (head_dim derived), trc:unverified -> needs-trust-remote-code-ack;
# --trust-remote-code -> engine-supported -> eligible -> estimated-lower-bound
# verdict, NO [D] (Path B).
s = "fixtures/llama-trc"
ffL = lambda: ff_derived(s, dense_cfg("LlamaForCausalLM"), weight_gb=8.0)
r = P.run_pull(s, "vllm/minimal", path="B", hardware_sm=SM_86,
fetcher=ffL(), profiles=profiles, statvfs=BIG_DISK)
check(r.stratum is P.Stratum.C0
and r.abort_reason == "needs-trust-remote-code-ack",
f"g4: Llama trc:unverified -> stratum-3 needs-trust-remote-code-ack "
f"(got {r.stratum.name}/{r.abort_reason})")
# + --trust-remote-code clears stratum-3; an estimated-lower-bound model
# then reaches [C1] = confirm→proceed which itself needs --yes (§4.1 — a
# low-confidence model is NEVER a silent pass). Without --yes it is an
# honest non-pass (NOT a fabricated fit).
r = P.run_pull(s, "vllm/minimal", path="B", hardware_sm=SM_86,
fetcher=ffL(), profiles=profiles, statvfs=BIG_DISK,
trust_remote_code=True)
check(not r.ok and r.stratum is P.Stratum.DECIDED
and r.confidence == "estimated-lower-bound"
and r.raw_verdict is not None and not r.emitted
and r.abort_reason.startswith("confirm→proceed"),
f"g4: + --trust-remote-code -> engine-supported -> [C1] "
f"confirm→proceed (needs --yes; honest non-pass, NO [D]) "
f"(ok={r.ok}, conf={r.confidence}, reason={r.abort_reason})")
# + --yes -> the Path-B verdict is download-eligible, still NO [D].
r = P.run_pull(s, "vllm/minimal", path="B", hardware_sm=SM_86,
fetcher=ffL(), profiles=profiles, statvfs=BIG_DISK,
trust_remote_code=True, yes=True)
check(r.ok and r.stratum is P.Stratum.DECIDED
and r.confidence == "estimated-lower-bound"
and r.raw_verdict is not None and not r.emitted,
f"g4: + --trust-remote-code --yes -> Path-B verdict, NO [D] "
f"(ok={r.ok}, conf={r.confidence}, emitted={r.emitted})")
check(any("soak-continuous" in n for n in r.notices),
"g4: Path-B verdict carries the §7 boot-fit≠runtime caveat")
# g5: no-arch-row; --experimental-arch -> eligible: verdict;
# ineligible -> stratum-5 no-fit-model (NOT bypassed) [also g11].
s = "fixtures/exotic-dense"
exo = dense_cfg("TotallyExoticForCausalLM")
r = P.run_pull(s, "vllm/minimal", path="B", hardware_sm=SM_86,
fetcher=ff_derived(s, exo), profiles=profiles,
statvfs=BIG_DISK)
check(r.stratum is P.Stratum.C0
and r.abort_reason == "engine-support-unknown/no-arch-row",
f"g5: exotic arch -> stratum-3 no-arch-row "
f"(got {r.stratum.name}/{r.abort_reason})")
# + --experimental-arch clears no-arch-row; eligible derived model reaches
# [C1] confirm→proceed (estimated-lower-bound) — needs --yes to be
# download-eligible.
r = P.run_pull(s, "vllm/minimal", path="B", hardware_sm=SM_86,
fetcher=ff_derived(s, exo), profiles=profiles,
statvfs=BIG_DISK, experimental_arch=True, yes=True)
check(r.ok and r.stratum is P.Stratum.DECIDED and r.raw_verdict is not None,
f"g5: + --experimental-arch --yes (eligible) -> Path-B verdict "
f"(ok={r.ok}, stratum={r.stratum.name})")
check(any("bypassed" in n for n in r.notices),
"g5: bypass surfaces a notice (Path B only, capture deferred)")
# g6a: known arch + auto_map -> --trust-remote-code only.
s = "fixtures/automap-llama"
am = dense_cfg("LlamaForCausalLM", auto_map={"x": "y"})
r = P.run_pull(s, "vllm/minimal", path="B", hardware_sm=SM_86,
fetcher=ff_derived(s, am), profiles=profiles,
statvfs=BIG_DISK)
check(r.stratum is P.Stratum.C0
and r.abort_reason == "needs-trust-remote-code-ack"
and r.diagnostics.get("c0_bypassable_by") == ["--trust-remote-code"],
f"g6a: known arch + auto_map -> needs-trc-ack, --trust-remote-code "
f"ONLY (got {r.diagnostics.get('c0_bypassable_by')})")
r = P.run_pull(s, "vllm/minimal", path="B", hardware_sm=SM_86,
fetcher=ff_derived(s, am), profiles=profiles,
statvfs=BIG_DISK, trust_remote_code=True, yes=True)
check(r.ok and r.stratum is P.Stratum.DECIDED,
"g6a: + --trust-remote-code --yes -> proceeds (Path-B verdict)")
# g6a: --experimental-arch alone does NOT clear an auto_map+known-arch trc.
r = P.run_pull(s, "vllm/minimal", path="B", hardware_sm=SM_86,
fetcher=ff_derived(s, am), profiles=profiles,
statvfs=BIG_DISK, experimental_arch=True)
check(r.stratum is P.Stratum.C0,
"g6a: --experimental-arch alone does NOT clear auto_map trc")
# g6b: no-arch-row + auto_map -> BOTH flags required.
s = "fixtures/automap-exotic"
amx = dense_cfg("ExoticAutoMapForCausalLM", auto_map={"x": "y"})
r = P.run_pull(s, "vllm/minimal", path="B", hardware_sm=SM_86,
fetcher=ff_derived(s, amx), profiles=profiles,
statvfs=BIG_DISK)
check(set(r.diagnostics.get("c0_bypassable_by") or [])
== {"--trust-remote-code", "--experimental-arch"},
f"g6b: no-arch-row+auto_map -> needs BOTH flags "
f"(got {r.diagnostics.get('c0_bypassable_by')})")
r = P.run_pull(s, "vllm/minimal", path="B", hardware_sm=SM_86,
fetcher=ff_derived(s, amx), profiles=profiles,
statvfs=BIG_DISK, trust_remote_code=True)
check(r.stratum is P.Stratum.C0,
"g6b: only --trust-remote-code (missing --experimental-arch) -> "
"still stratum-3 (subset rule)")
r = P.run_pull(s, "vllm/minimal", path="B", hardware_sm=SM_86,
fetcher=ff_derived(s, amx), profiles=profiles,
statvfs=BIG_DISK, trust_remote_code=True,
experimental_arch=True, yes=True)
check(r.ok and r.stratum is P.Stratum.DECIDED,
"g6b: BOTH flags (+ --yes) -> clears stratum-3 -> Path-B verdict")
# g10: runtime-incompatible is NON-bypassable; --experimental-arch does
# NOT bypass it. Curated MoE arch with a loads:false pin (Path B, arch from
# config) -> runtime-incompatible.
s = "fixtures/qwen35-moe"
moe = {
"model_type": "qwen3_5_moe",
"architectures": ["Qwen3_5MoeForConditionalGeneration"],
"hidden_size": 4096, "num_hidden_layers": 32,
"num_attention_heads": 32, "num_key_value_heads": 8,
"num_local_experts": 128, "torch_dtype": "bfloat16",
}
r = P.run_pull(s, "vllm/default", path="B", hardware_sm=SM_90,
fetcher=ff_derived(s, moe), profiles=profiles,
statvfs=BIG_DISK)
check(r.stratum is P.Stratum.C0
and r.abort_reason == "engine-support-unknown/runtime-incompatible",
f"g10: loads:false pin -> stratum-3 runtime-incompatible "
f"(got {r.stratum.name}/{r.abort_reason})")
r = P.run_pull(s, "vllm/default", path="B", hardware_sm=SM_90,
fetcher=ff_derived(s, moe), profiles=profiles,
statvfs=BIG_DISK, experimental_arch=True)
check(r.stratum is P.Stratum.C0
and r.abort_reason == "engine-support-unknown/runtime-incompatible",
"g10: --experimental-arch does NOT bypass runtime-incompatible "
"(non-bypassable)")
# g14: hardware-SM mismatch. Gemma fp8_e4m3 / required_sm 9.0 profile on a
# detected RTX 3090 (sm_86) -> runtime-incompatible (no false 'fits'),
# non-bypassable. (Curated Gemma 31B + vllm/gemma-mtp-tp1.)
GEMMA_SLUG = "Intel/gemma-4-31B-it-int4-AutoRound"
r = P.run_pull(GEMMA_SLUG, "vllm/gemma-mtp-tp1", path="A",
hardware_sm=SM_86, fetcher=NoNet(), profiles=profiles,
statvfs=BIG_DISK)
check(r.stratum is P.Stratum.C0
and r.abort_reason == "engine-support-unknown/runtime-incompatible",
f"g14: fp8_e4m3/required_sm 9.0 on sm_86 -> stratum-3 "
f"runtime-incompatible (got {r.stratum.name}/{r.abort_reason})")
r = P.run_pull(GEMMA_SLUG, "vllm/gemma-mtp-tp1", path="A",
hardware_sm=SM_86, fetcher=NoNet(), profiles=profiles,
statvfs=BIG_DISK, experimental_arch=True)
check(r.stratum is P.Stratum.C0,
"g14: --experimental-arch does NOT bypass the SM mismatch "
"(non-bypassable)")
# Same profile on sm_90 clears the SM gate (proves it IS the SM gate).
r = P.run_pull(GEMMA_SLUG, "vllm/gemma-mtp-tp1", path="A",
hardware_sm=SM_90, fetcher=NoNet(), profiles=profiles,
statvfs=BIG_DISK)
check(r.stratum is not P.Stratum.C0
or r.abort_reason != "engine-support-unknown/runtime-incompatible",
"g14: same profile on sm_90 passes [C0] SM gate (proves SM gate)")
# ===========================================================================
# SECTION 5 — STRATUM 4: [C2a] disk pre-gate (g7) + ordering AFTER [C0].
# ===========================================================================
print("\n--- stratum-4: [C2a] disk (g7) + order vs [C0] ---")
# g7: disk-short -> stratum-4 hard-abort AFTER [C0], BEFORE [B].
# (--trust-remote-code clears the trc:unverified [C0] so the run REACHES
# stratum-4 — proving [C2a] is evaluated AFTER a clean [C0].)
s = "fixtures/big-llama"
r = P.run_pull(s, "vllm/minimal", path="B", hardware_sm=SM_86,
fetcher=ff_derived(s, dense_cfg("Qwen2ForCausalLM"),
weight_gb=200.0),
profiles=profiles, statvfs=TINY_DISK,
trust_remote_code=True)
check(r.stratum is P.Stratum.C2A_DISK and r.abort_reason == "disk-short",
f"g7: 200 GiB model, 2 GiB free, [C0] clean -> stratum-4 "
f"disk-short (got {r.stratum.name}/{r.abort_reason})")
# Ordering: a CLEAN [C0] still hard-aborts on disk-short, AND [B] never ran
# (no fit verdict).
check(r.raw_verdict is None,
"g7 ordering: [C2a] is AFTER [C0] and BEFORE [B] "
"(no fit verdict produced before the disk abort)")
# A no-arch-row model with disk-short stops at stratum-3 FIRST (proves
# strict 3-before-4 ordering: [C0] precedes [C2a]).
s = "fixtures/exotic-bigdisk-short"
r = P.run_pull(s, "vllm/minimal", path="B", hardware_sm=SM_86,
fetcher=ff_derived(s, dense_cfg("ZQZForCausalLM"),
weight_gb=200.0),
profiles=profiles, statvfs=TINY_DISK)
check(r.stratum is P.Stratum.C0,
"ordering: no-arch-row + disk-short stops at stratum-3 FIRST "
"([C0] strictly precedes [C2a])")
# ===========================================================================
# SECTION 6 — STRATUM 5: pre-[B] no-fit-model (g11) — monotonic,
# non-bypassable; [C0] stays engine-supported.
# ===========================================================================
print("\n--- stratum-5: pre-[B] no-fit-model (g11) ---")
# g11: Gemma2-SWA-only style ineligible derived model with a KNOWN arch row
# (so [C0]=engine-supported, stays) -> stratum-5 no-fit-model.
# Use a SWA-only config on a no-arch-row arch + --experimental-arch so [C0]
# resolves (engine-support bypassed) then stratum-5 STILL fires (proves
# --experimental-arch does NOT bypass stratum-5).
s = "fixtures/swa-only"
swa = dense_cfg("SwaOnlyForCausalLM", sliding_window=4096)
swa.pop("max_window_layers", None)
r = P.run_pull(s, "vllm/minimal", path="B", hardware_sm=SM_86,
fetcher=ff_derived(s, swa), profiles=profiles,
statvfs=BIG_DISK, experimental_arch=True)
check(r.stratum is P.Stratum.ELIGIBILITY
and r.abort_reason == "no-fit-model",
f"g11: SWA-only ineligible -> stratum-5 no-fit-model "
f"(got {r.stratum.name}/{r.abort_reason})")
check("non-bypassable" in r.detail,
"g11: no-fit-model is non-bypassable (detail states it)")
# MoE-on-known-arch path: [C0]=engine-supported (stays) -> stratum-5.
# Mixtral has no arch row; use a config that is MoE AND has the curated
# Qwen MoE arch row (engine-supported on a clean engine) to prove [C0] is
# NOT rewritten by stratum-5.
s = "fixtures/moe-known"
moe2 = {
"model_type": "qwen3_5_moe",
"architectures": ["Qwen3_5MoeForConditionalGeneration"],
"hidden_size": 4096, "num_hidden_layers": 32,
"num_attention_heads": 32, "num_key_value_heads": 8,
"num_local_experts": 128, "torch_dtype": "bfloat16",
}
c0 = G.c0_engine_support(
"vllm/qwen-a3b-preview", P.D.DeriveResult(
slug=s, profile={"arch": "Qwen3_5MoeForConditionalGeneration",
"auto_map": False}),
path="B", hardware_sm=SM_86, root=root)
r = P.run_pull(s, "vllm/qwen-a3b-preview", path="B", hardware_sm=SM_86,
fetcher=ff_derived(s, moe2), profiles=profiles,
statvfs=BIG_DISK, experimental_arch=True)
check(c0.state is G.C0State.ENGINE_SUPPORTED
and r.stratum is P.Stratum.ELIGIBILITY
and r.abort_reason == "no-fit-model",
f"g11 monotonic: [C0]=engine-supported (stays) -> stratum-5 "
f"no-fit-model ([C0]={c0.state.value}, "
f"got {r.stratum.name}/{r.abort_reason})")
# --experimental-arch did NOT bypass stratum-5 (it ran WITH the flag set).
check(r.abort_reason == "no-fit-model",
"g11: --experimental-arch does NOT bypass stratum-5 no-fit-model")
# ===========================================================================
# SECTION 7 — STRATUM 6: Path-A [D] dry-run (g1, g2, g8, g15) + Path-B
# never touches [D].
# ===========================================================================
print("\n--- stratum-6: Path-A [D] dry-run + Path-B isolation "
"(g1/g2/g8/g15) ---")
# g8: curated + --dry-run -> Path-B verdict, NO [D] (forced Path B even
# though the slug is curated and an --out is given).
r = P.run_pull(CURATED_SLUG, "vllm/dual", dry_run=True, out="/tmp/_no.yml",
hardware_sm=SM_86, fetcher=NoNet(), profiles=profiles,
statvfs=BIG_DISK, yes=True)
check(r.path == "B" and not r.emitted and r.compose_text is None,
f"g8: curated + --dry-run -> Path B, NO [D] emit "
f"(path={r.path}, emitted={r.emitted})")
check(not os.path.exists("/tmp/_no.yml"),
"g8: Path B wrote NO compose file even with --out (never emits)")
# g2: Path-A curated `vllm/dual` -> [C1] terminal is whatever the
# AUTHORITATIVE curated-exact kv-calc spec prices it (parameterized — NOT
# hardcoded; can never silently diverge from kv-calc). When the curated-exact
# verdict is download-eligible, --yes (when the §4.1 row needs it) -> [D]
# dry-run -> emit; the non-silent rows still require --yes.
OUT = "/tmp/_pull_g2.yml"
if os.path.exists(OUT):
os.unlink(OUT)
g2_rv, g2_term = curated_exact_expectation("vllm/dual")
g2_silent = (g2_term == "proceed") # only exact×fits-clean is silent
r = P.run_pull(CURATED_SLUG, "vllm/dual", path="A", out=OUT,
hardware_sm=SM_86, fetcher=NoNet(), profiles=profiles,
statvfs=BIG_DISK, yes=True)
check(r.ok and r.path == "A" and r.raw_verdict == g2_rv
and r.terminal == g2_term and r.emitted,
f"g2: Path-A curated vllm/dual + --yes -> kv-calc curated-exact "
f"verdict={g2_rv} terminal={g2_term} -> [D] emit "
f"(ok={r.ok}, verdict={r.raw_verdict}, terminal={r.terminal}, "
f"emitted={r.emitted})")
if g2_rv == "fits-constrained":
# fits-constrained invariant: surfaces the known effective-cap warning
# (NOT 'applied constraint'; no rewrite). Only assertable when kv-calc
# actually prices this curated config constrained.
cap_notice = [n for n in r.notices if "effective-cap warning" in n]
check(cap_notice and "no compose config rewritten" in cap_notice[0]
and "applied constraint" not in cap_notice[0],
"g2: fits-constrained prints 'known effective-cap warning' "
"(NOT 'applied constraint'; no rewrite)")
# without --yes: a non-silent terminal is NOT satisfied (needs --yes); a
# silent exact×fits-clean `proceed` IS satisfied even without --yes.
r2 = P.run_pull(CURATED_SLUG, "vllm/dual", path="A", out=OUT + ".x",
hardware_sm=SM_86, fetcher=NoNet(), profiles=profiles,
statvfs=BIG_DISK)
if g2_silent:
check(r2.ok and r2.emitted and r2.terminal == "proceed",
f"g2: Path-A curated vllm/dual silent {g2_term} -> eligible "
f"WITHOUT --yes (ok={r2.ok}, emitted={r2.emitted})")
check(os.path.exists(OUT + ".x"),
"g2: a satisfied silent proceed wrote the compose file")
os.unlink(OUT + ".x")
else:
check(not r2.ok and not r2.emitted
and r2.abort_reason.startswith(g2_term),
f"g2: Path-A curated vllm/dual non-silent {g2_term} WITHOUT "
f"--yes -> NOT eligible, NO [D] (reason={r2.abort_reason})")
check(not os.path.exists(OUT + ".x"),
"g2: a non-satisfied non-silent terminal emits NO compose file")
# g1: Path-A curated, gate-passing, [D] dry-run ok -> proceed → [D] invoked.
# (We use the satisfied confirm→proceed above as the integration analogue
# of g1; here also assert the emitted body + that --project-directory note
# is surfaced, and the trc-leak invariant.)
check(r.compose_text is not None and "services:" in r.compose_text,
"g1/g2: [D] produced a compose body on a clean dry-run")
check(os.path.exists(OUT),
"g1/g2: Path-A satisfied terminal wrote the compose to --out")
check(any("--project-directory" in n for n in r.notices),
"g1/g2: emitted compose carries the COMPOSE_GENERATOR.md "
"--project-directory consumption note")
# trc-leak invariant: no --trust-remote-code in the in-scope Path-A emitted
# body (reuse the patch_attribution.service_body()==0 check pattern).
body = PA.service_body(r.compose_text)
check(body.count("--trust-remote-code") == 0,
"g1/g2: in-scope Path-A emitted body contains ZERO "
"--trust-remote-code (governed-slot invariant, service_body()==0)")
# g15: Path-A passes stratum-2 + [C1] download-eligible but the [D] dry-run
# REFUSES at a later point (foundational/degraded patch drift) -> Path-A
# abort, NOT reported download-eligible. Inject a d_runner that refuses.
def refusing_runner(_root, _profile, _ad):
raise GC.Refuse("PN12: foundational drift-guard failed (compile-safe) "
"-> hard-refuse")
r = P.run_pull(CURATED_SLUG, "vllm/dual", path="A", out=OUT + ".g15",
hardware_sm=SM_86, fetcher=NoNet(), profiles=profiles,
statvfs=BIG_DISK, yes=True, d_runner=refusing_runner)
check(r.stratum is P.Stratum.D_DRY_RUN
and r.abort_reason == "d-refused:foundational-drift"
and not r.ok and not r.emitted,
f"g15: [D] dry-run refuses -> stratum-6 Path-A abort, NOT "
f"download-eligible (got {r.stratum.name}/{r.abort_reason}, "
f"ok={r.ok})")
check(not os.path.exists(OUT + ".g15"),
"g15: a [D]-refused dry-run wrote NO compose file")
# g3: Path-A curated `vllm/minimal` -> [C1] terminal is whatever the
# AUTHORITATIVE curated-exact kv-calc spec prices it (parameterized — NOT
# hardcoded; can never silently freeze a misprice again). The structural
# invariant under test: exact×wont-fit is an UNCLEARABLE hard-block (no flag
# clears it, no [D]); a download-eligible curated-exact verdict emits.
g3_rv, g3_term = curated_exact_expectation("vllm/minimal")
r = P.run_pull(CURATED_SLUG, "vllm/minimal", path="A", out=OUT + ".g3",
hardware_sm=SM_86, fetcher=NoNet(), profiles=profiles,
statvfs=BIG_DISK, yes=True, force_download=True)
check(r.terminal == g3_term,
f"g3: Path-A curated vllm/minimal terminal == kv-calc curated-exact "
f"({g3_rv} -> {g3_term}); got {r.terminal}")
if g3_rv == "wont-fit":
check(not r.ok and not r.emitted and r.abort_reason == "hard-block",
f"g3: exact×wont-fit -> hard-block UNCLEARABLE even with --yes "
f"--force-download, NO [D] (ok={r.ok}, "
f"reason={r.abort_reason})")
check(not os.path.exists(OUT + ".g3"),
"g3: hard-block wrote NO compose file")
else:
check(r.ok and r.emitted,
f"g3: curated-exact download-eligible ({g3_term}) -> [D] emit "
f"(ok={r.ok}, emitted={r.emitted})")
check(os.path.exists(OUT + ".g3"),
"g3: a download-eligible curated terminal wrote the compose")
# ---------------------------------------------------------------------------
# P4-fix REGRESSION (NON-MOCKED): Tier-1 curated hit must be priced through
# the model's authoritative curated-exact kv-calc family branch, NOT P1's
# conservative generic-dense lower-bound.
#
# This drives the REAL `run_pull` -> REAL `_curated_spec` -> REAL
# `kv.raw_verdict` (real kv-calc) for a Tier-1 curated `(slug, profile-like)`
# pair (no FixtureFetcher for the verdict; NoNet proves it stays network-free)
# and asserts the Pull-Gate `[B]` raw_verdict + `[C1]` terminal EQUAL kv-calc's
# OWN curated-exact `predict()` mapped through FAIL/TIGHT/PASS ->
# wont-fit/fits-constrained/fits-clean. Pre-fix: `_curated_spec` emitted
# model_family="generic-dense" -> wont-fit -> false `hard-block` (no compose).
# Expectation is parameterized off `kv.predict()` + `kv.MODEL_SPECS[...]`, so
# the gate can NEVER silently diverge from kv-calc again. Both curated vLLM
# profiles for this model FIT under curated-exact pricing -> NOT a false
# hard-block (the proven symptom this fix closes).
for _pl in ("vllm/minimal", "vllm/dual"):
exp_rv, exp_term = curated_exact_expectation(_pl)
_o = f"/tmp/_pull_p4fix_{_pl.replace('/', '_')}.yml"
if os.path.exists(_o):
os.unlink(_o)
rr = P.run_pull(CURATED_SLUG, _pl, path="A", out=_o,
hardware_sm=SM_86, fetcher=NoNet(), profiles=profiles,
statvfs=BIG_DISK, yes=True)
check(rr.raw_verdict == exp_rv,
f"P4-fix[{_pl}]: Pull-Gate [B] raw_verdict == kv-calc "
f"curated-exact ({exp_rv}); got {rr.raw_verdict} "
f"(curated model NOT priced as generic-dense)")
check(rr.terminal == exp_term,
f"P4-fix[{_pl}]: [C1] terminal == kv-calc-derived exact-row "
f"terminal ({exp_term}); got {rr.terminal}")
if exp_rv != "wont-fit":
check(rr.terminal != "hard-block" and rr.ok and rr.emitted
and os.path.exists(_o),
f"P4-fix[{_pl}]: kv-calc PASS/TIGHT -> NOT a false "
f"hard-block; curated download-eligible + compose emitted "
f"(terminal={rr.terminal}, ok={rr.ok}, emitted={rr.emitted})")
if os.path.exists(_o):
os.unlink(_o)
# Path-B structural isolation: NO Path-B run can ever set emitted/
# compose_text. Sweep a representative matrix.
pathb_emit = False
for pl in ("vllm/minimal", "vllm/dual", "vllm/tools-text"):
rb = P.run_pull(CURATED_SLUG, pl, dry_run=True, hardware_sm=SM_86,
fetcher=NoNet(), profiles=profiles, statvfs=BIG_DISK,
yes=True, force_download=True)
if rb.emitted or rb.compose_text is not None:
pathb_emit = True
check(not pathb_emit,
"Path B is structurally incapable of [D] emit / download "
"(--force-download is a no-op + notice this phase)")
# g9: generic-dense lower-bound is honored end-to-end + --calibration 22/22
# is asserted by the harness below (kv-calc unchanged).
s = "fixtures/llama-dense-elb"
r = P.run_pull(s, "vllm/minimal", path="B", hardware_sm=SM_86,
fetcher=ff_derived(s, dense_cfg("Qwen2ForCausalLM"),
weight_gb=4.0),
profiles=profiles, statvfs=BIG_DISK, yes=True,
trust_remote_code=True)
check(r.confidence == "estimated-lower-bound" and r.raw_verdict is not None,
f"g9: derived dense -> estimated-lower-bound confidence, real "
f"verdict (conf={r.confidence}, verdict={r.raw_verdict})")
# ===========================================================================
# SECTION 8 — STRATUM ORDERING PROOF (1→2→3→4→5→6 strict & monotonic).
# Stack multiple failing conditions; assert the EARLIEST stratum wins.
# ===========================================================================
print("\n--- stratum ordering proof (earliest-wins, monotonic) ---")
# stratum-1 BEFORE stratum-2: a deriver error + a bad profile-like ->
# stratum-1 wins (deriver runs first).
s = "fixtures/order-s1"
ff = FixtureFetcher({API.format(slug=s): D.FetchResponse(404, b"")})
r = P.run_pull(s, "llamacpp/default", hardware_sm=SM_86, fetcher=ff,
profiles=profiles, statvfs=TINY_DISK)
check(r.stratum is P.Stratum.DERIVER,
"order: stratum-1 (deriver) precedes stratum-2 (bad profile-like) "
"+ stratum-4 (tiny disk)")
# stratum-2 BEFORE stratum-3: bad profile-like + would-be-bad [C0] ->
# stratum-2 wins.
s = "fixtures/order-s2"
r = P.run_pull(s, "llamacpp/default", path="B", hardware_sm=SM_86,
fetcher=ff_derived(s, dense_cfg("ExoticXForCausalLM")),
profiles=profiles, statvfs=TINY_DISK)
check(r.stratum is P.Stratum.PROFILE_LIKE,
"order: stratum-2 (profile-like) precedes stratum-3 ([C0]) + "
"stratum-4 (disk)")
# stratum-3 BEFORE stratum-4: no-arch-row + disk-short -> stratum-3 wins.
s = "fixtures/order-s3"
r = P.run_pull(s, "vllm/minimal", path="B", hardware_sm=SM_86,
fetcher=ff_derived(s, dense_cfg("NoRowForCausalLM"),
weight_gb=200.0),
profiles=profiles, statvfs=TINY_DISK)
check(r.stratum is P.Stratum.C0,
"order: stratum-3 ([C0]) precedes stratum-4 ([C2a] disk)")
# stratum-4 BEFORE stratum-5: disk-short + ineligible (MoE) on a clean
# [C0] -> stratum-4 wins.
s = "fixtures/order-s4"
moec = {
"model_type": "qwen3_5_moe",
"architectures": ["Qwen3_5MoeForConditionalGeneration"],
"hidden_size": 4096, "num_hidden_layers": 32,
"num_attention_heads": 32, "num_key_value_heads": 8,
"num_local_experts": 128, "torch_dtype": "bfloat16",
}
r = P.run_pull(s, "vllm/qwen-a3b-preview", path="B", hardware_sm=SM_86,
fetcher=ff_derived(s, moec, weight_gb=300.0),
profiles=profiles, statvfs=TINY_DISK,
experimental_arch=True)
check(r.stratum is P.Stratum.C2A_DISK,
"order: stratum-4 ([C2a] disk-short) precedes stratum-5 "
"(no-fit-model)")
# stratum-5 BEFORE [B]/[C1]: ineligible model never reaches a verdict.
s = "fixtures/order-s5"
r = P.run_pull(s, "vllm/qwen-a3b-preview", path="B", hardware_sm=SM_86,
fetcher=ff_derived(s, moec, weight_gb=4.0),
profiles=profiles, statvfs=BIG_DISK,
experimental_arch=True)
check(r.stratum is P.Stratum.ELIGIBILITY and r.raw_verdict is None,
"order: stratum-5 (no-fit-model) precedes [B] (no fit verdict "
"produced)")
# stratum-6 is Path-A-only: the same gate-passing model on Path B NEVER
# reaches stratum-6 (no [D] dry-run on Path B).
r = P.run_pull(CURATED_SLUG, "vllm/dual", dry_run=True, hardware_sm=SM_86,
fetcher=NoNet(), profiles=profiles, statvfs=BIG_DISK,
yes=True)
check(r.stratum is not P.Stratum.D_DRY_RUN and not r.emitted,
"order: stratum-6 ([D] dry-run) is Path-A ONLY (Path B never "
"reaches it)")
# --experimental-arch bypass scoping — consolidated assertion:
# bypasses no-arch-row; does NOT bypass runtime-incompatible; does NOT
# bypass stratum-5 no-fit-model. (Cited individually in g5/g10/g14/g11.)
s = "fixtures/scope-norow"
r_norow = P.run_pull(s, "vllm/minimal", path="B", hardware_sm=SM_86,
fetcher=ff_derived(s, dense_cfg("ScopeXForCausalLM")),
profiles=profiles, statvfs=BIG_DISK,
experimental_arch=True)
check(r_norow.ok or r_norow.stratum is not P.Stratum.C0,
"scope: --experimental-arch BYPASSES no-arch-row")
check(
P.run_pull("fixtures/qwen35-moe2", "vllm/default", path="B",
hardware_sm=SM_90,
fetcher=ff_derived("fixtures/qwen35-moe2", {
"model_type": "qwen3_5_moe",
"architectures": ["Qwen3_5MoeForConditionalGeneration"],
"hidden_size": 4096, "num_hidden_layers": 32,
"num_attention_heads": 32, "num_key_value_heads": 8,
"num_local_experts": 128, "torch_dtype": "bfloat16"}),
profiles=profiles, statvfs=BIG_DISK,
experimental_arch=True).abort_reason
== "engine-support-unknown/runtime-incompatible",
"scope: --experimental-arch does NOT bypass runtime-incompatible",
)
# ===========================================================================
# SECTION 9 — v0.8.0 [E] E4: post-[C1] derived-[E] orchestration + trigger
# semantics + override force-capture (pt5). All [E]-stage funcs MOCKED
# (NO real Docker / GPU / network — real on-rig is E5). These are ADDED
# cases (g16..g22); every assertion above (§4.1 9-cell, 6-stratum, g0..g15,
# trc-leak, Path-B isolation) is unchanged and stays green.
# ===========================================================================
print("\n--- [E] E4: post-[C1] derived orchestration + triggers "
"(g16..g22, mocked) ---")
import shutil as _sh # noqa: E402
from scripts.lib.profiles import capture as _CAP # noqa: E402
# E4 [E]-stage capture artifacts land under <root>/.pull-captures (the
# real runtime dir); these are mocked-run artifacts — purge the whole tree
# at the end so the test leaves NO repo residue (and it is .gitignore'd).
_CAP_ROOT = root / ".pull-captures"
def _purge_captures():
_sh.rmtree(_CAP_ROOT, ignore_errors=True)
# Injected GPU topology (no real nvidia-smi dependence): 2× 24 GiB.
TOPO_2 = (2, [24576, 24576], ["NVIDIA GeForce RTX 3090",
"NVIDIA GeForce RTX 3090"])
TOPO_1 = (1, [24576], ["NVIDIA GeForce RTX 3090"])
class _Calls:
def __init__(self):
self.emit = self.dl = self.boot = self.smoke = 0
self.cap = self.p5 = 0
# E3/E4-fix ordering ledger: every lifecycle event appended in
# invocation order so a test can PROVE the new contract —
# boot-up -> (server ALIVE) smoke -> capture [-> pt5] -> teardown
# i.e. teardown index > capture index > smoke index > boot index,
# and smoke/capture see `server_alive == True`.
self.order: list[str] = []
self.server_alive = False
self.smoke_saw_alive = None
self.capture_saw_alive = None
def mk_emocks(calls, *, emittable=True, boot_ok=True):
"""Mocked E1/E2/E3 + E4-pt5 funcs. NO Docker/GPU/network."""
from scripts.lib.profiles.downloader import DownloadResult
from scripts.lib.profiles.booter import BootResult
def emit_fn(root, ei):
calls.emit += 1
if not emittable:
raise GC.Refuse("derived-runtime-unsupported:kv")
return ("services:\n vllm-derived-x:\n image: img:pin\n",
{"resolved_image": "img:pin", "max_model_len": 32768,
"kv_format": ei.runtime.get("kv_format"),
"engine": ei.runtime.get("engine")})
def download_fn(ei, fetcher=None):
calls.dl += 1
return DownloadResult(ok=True, files=["model.safetensors"],
bytes=8_000_000_000, sha_verified=True,
failure=None,
local_dir=str(ei.hf_home))
import contextlib as _cl
@_cl.contextmanager
def boot_cm(ei, compose_text):
# E3/E4-fix: the boot lifecycle is a CONTEXT MANAGER. up() FIRST,
# then `yield` the live handle WHILE the (fixture) server is "up";
# teardown (down) ALWAYS in the finally, on __exit__, AFTER the
# with-body (smoke + capture + pt5). This fixture records the exact
# event order so the test can prove up -> smoke -> capture -> down
# and that teardown still runs if the with-body raises.
calls.boot += 1
calls.order.append("up")
calls.server_alive = True
try:
if boot_ok:
yield BootResult(ok=True, seconds=1.0, failure=None,
endpoint="http://127.0.0.1:8020/v1")
else:
yield BootResult(ok=False, seconds=0.5,
failure="CUDA OOM; worker exited",
endpoint=None)
finally:
# ALWAYS — even if the with-body raised (no-orphan guarantee at
# the CORRECT scope: AFTER smoke + capture, never before).
calls.server_alive = False
calls.order.append("down")
def smoke_fn(ei, endpoint):
calls.smoke += 1
calls.order.append("smoke")
# PROOF the server is ALIVE when smoke runs (the whole point of the
# E3/E4-fix — the prior teardown-in-finally-before-return killed it).
calls.smoke_saw_alive = calls.server_alive
return _CAP.SmokeResult(
smoke_capability_set=["plain-chat", "streaming"],
results={"plain-chat": "green", "streaming": "green",
"tool-call": "unsmoked", "reasoning-streaming":
"unsmoked", "structured-output": "unsmoked",
"vision": "unsmoked", "long-context": "unsmoked"},
partial=True)
def capture_fn(ei, **kw):
calls.cap += 1
calls.order.append("capture")
# Capture is also emitted while the server is still UP (before the
# CM __exit__ teardown).
calls.capture_saw_alive = calls.server_alive
return _CAP.emit_capture(ei, **kw)
def override_capture_fn(ei, **kw):
calls.p5 += 1
calls.order.append("pt5")
return _CAP.emit_override_capture(ei, **kw)
return dict(emit_fn=emit_fn, download_fn=download_fn, boot_cm=boot_cm,
smoke_fn=smoke_fn, capture_fn=capture_fn,
override_capture_fn=override_capture_fn)
import tempfile as _tf # noqa: E402
# A non-curated generic-dense Llama (Qwen2 arch -> known arch row, no trc).
# torch_dtype bfloat16 -> weight_format bfloat16 (pure-dtype CONTRACT-2 row;
# CONTRACT-5 dispatch passes). vllm/minimal: clean engine, fp8_e5m2 KV,
# drafter None, no required features, tp=1 -> derived-emittable.
DSLUG = "fixtures/e4-dense"
def derived_run(profile_like, *, calls, emittable=True, boot_ok=True,
topo=TOPO_2, **kw):
with _tf.TemporaryDirectory() as _td:
# repo_root for capture goes to a tmp dir via root override is not
# possible (root selects the registry); capture writes under
# <root>/.pull-captures — use the real root but a unique slug+ts so
# artifacts are isolated. We assert on res.capture_paths.
return P.run_pull(
DSLUG, profile_like, path="B", hardware_sm=SM_86,
fetcher=ff_derived(DSLUG, dense_cfg("Qwen2ForCausalLM"),
weight_gb=4.0),
profiles=profiles, statvfs=BIG_DISK, trust_remote_code=True,
gpu_topology=topo, **mk_emocks(calls, emittable=emittable,
boot_ok=boot_ok), **kw)
# --- g16: non-curated proceed (download-eligible), NO --dry-run -> [E] ----
c = _Calls()
r = derived_run("vllm/minimal", calls=c, yes=True)
check(r.ok and not r.emitted and r.confidence == "estimated-lower-bound",
f"g16: non-curated download-eligible verdict stands "
f"(ok={r.ok}, emitted={r.emitted}, conf={r.confidence})")
check(c.emit == 1 and c.dl == 1 and c.boot == 1 and c.cap == 1,
f"g16: [E] ran emit+download+boot+capture "
f"(emit={c.emit} dl={c.dl} boot={c.boot} cap={c.cap})")
check(r.download_ok is True and r.boot_ok is True
and isinstance(r.smoke, dict)
and r.smoke["smoke_capability_set"] == ["plain-chat", "streaming"],
f"g16: PullResult [E] additive fields populated "
f"(dl_ok={r.download_ok} boot_ok={r.boot_ok} smoke={r.smoke})")
_g16caps = [x for x in r.capture_paths if x.endswith(".json")]
check(any("pt1-gate" in x for x in _g16caps)
and any("pt2-download" in x for x in _g16caps)
and any("pt3-boot" in x for x in _g16caps)
and any("pt4-smoke" in x for x in _g16caps)
and any("manifest" in x for x in _g16caps)
and not any("pt5" in x or "override" in x for x in _g16caps),
f"g16: pt1-4 + manifest emitted, NO pt5 (non-override) "
f"(paths={[os.path.basename(x) for x in _g16caps]})")
check(c.p5 == 0, "g16: override-capture (pt5) NOT invoked (not override)")
# --- g16b: E3/E4-fix — boot lifecycle is a context manager; the server
# stays UP for smoke+capture; teardown happens on CM __exit__ AFTER
# the with-body. This is the assertion that PROVES the on-rig E5
# teardown-in-finally-before-smoke defect is fixed: prior code tore the
# container down BEFORE smoke -> ConnectionRefused, by construction. ---
_oi = {ev: i for i, ev in enumerate(c.order)}
check(c.order == ["up", "smoke", "capture", "down"],
f"g16b: lifecycle order is up -> smoke -> capture -> down EXACTLY "
f"(got {c.order})")
check(_oi["up"] < _oi["smoke"] < _oi["capture"] < _oi["down"],
f"g16b: teardown idx > capture idx > smoke idx > boot/up idx "
f"(order={c.order})")
check(c.smoke_saw_alive is True,
"g16b: smoke ran while the server was ALIVE (NOT torn down first — "
"the on-rig E5 ConnectionRefused root-cause is fixed)")
check(c.capture_saw_alive is True,
"g16b: capture (pt1-4 + manifest) emitted while the server was still "
"UP, BEFORE the CM __exit__ teardown")
check(c.server_alive is False,
"g16b: after run_pull returned the CM __exit__ ran -> server torn "
"down (no-orphan guarantee preserved at the CORRECT scope)")
for _x in r.capture_paths:
try:
os.unlink(_x)
except OSError:
pass
# --- g16c: E3/E4-fix — teardown STILL runs (no-orphan) even if capture
# RAISES inside the `with` body. The CM's finally must fire on the
# exceptional __exit__; run_pull must NOT raise (PullResult intact). --
c = _Calls()
def _boom_capture(ei, **kw):
c.cap += 1
c.order.append("capture")
c.capture_saw_alive = c.server_alive
raise RuntimeError("capture blew up mid-with-body")
_em = mk_emocks(c, emittable=True, boot_ok=True)
_em["capture_fn"] = _boom_capture
r = P.run_pull(
DSLUG, "vllm/minimal", path="B", hardware_sm=SM_86,
fetcher=ff_derived(DSLUG, dense_cfg("Qwen2ForCausalLM"), weight_gb=4.0),
profiles=profiles, statvfs=BIG_DISK, trust_remote_code=True,
yes=True, gpu_topology=TOPO_2, **_em)
check(isinstance(r, P.PullResult) and r.ok,
f"g16c: capture raising inside the `with` does NOT raise out of "
f"run_pull; PullResult intact (ok={r.ok})")
check("down" in c.order and c.order.index("down") > c.order.index("capture")
and c.server_alive is False,
f"g16c: CM teardown STILL ran on the exceptional __exit__, AFTER the "
f"capture that raised (no-orphan preserved) (order={c.order})")
check(c.capture_saw_alive is True,
"g16c: the raising capture still saw the server ALIVE (smoke+capture "
"scope is correct even on the failure path)")
check(any("capture emit failed" in n for n in r.notices),
"g16c: the capture exception is recorded structurally as a notice")
_purge_captures()
# --- g17: non-curated + --dry-run -> verdict-only, NO [E] ----------------
c = _Calls()
r = P.run_pull(DSLUG, "vllm/minimal", dry_run=True, hardware_sm=SM_86,
fetcher=ff_derived(DSLUG, dense_cfg("Qwen2ForCausalLM"),
weight_gb=4.0),
profiles=profiles, statvfs=BIG_DISK, trust_remote_code=True,
yes=True, gpu_topology=TOPO_2, **mk_emocks(c))
check(r.path == "B" and (c.emit + c.dl + c.boot + c.cap + c.p5) == 0,
f"g17: --dry-run stays verdict-only — NO [E] stage at all "
f"(emit={c.emit} dl={c.dl} boot={c.boot} cap={c.cap} p5={c.p5})")
check(r.download_ok is None and r.boot_ok is None and r.smoke is None
and r.capture_paths == [],
f"g17: --dry-run leaves all [E] additive fields None/empty "
f"(dl_ok={r.download_ok} boot_ok={r.boot_ok})")
check(any("soak-continuous" in n for n in r.notices),
"g17: --dry-run verdict still carries the §7 caveat (unchanged)")
# --- g18: override-accepted + --force-download -> [E] + pt5 emitted ------
# estimated-lower-bound × wont-fit + --force-download -> override-accepted.
# A huge-weight derived model forces wont-fit on this hardware.
c = _Calls()
r = P.run_pull(DSLUG, "vllm/minimal", path="B", hardware_sm=SM_86,
fetcher=ff_derived(DSLUG, dense_cfg("Qwen2ForCausalLM"),
weight_gb=400.0),
profiles=profiles, statvfs=BIG_DISK, trust_remote_code=True,
force_download=True, gpu_topology=TOPO_2,
**mk_emocks(c, boot_ok=False))
check(r.ok and r.terminal == "override-accepted",
f"g18: derived wont-fit + --force-download -> override-accepted "
f"(ok={r.ok}, terminal={r.terminal})")
check(c.emit == 1 and c.dl == 1 and c.boot == 1 and c.cap == 1
and c.p5 == 1,
f"g18: --force-download ACTIVATES [E] for override-accepted incl "
f"pt5 (emit={c.emit} dl={c.dl} boot={c.boot} cap={c.cap} p5={c.p5})")
_p5 = [x for x in r.capture_paths if "pt5" in x or "override" in x]
check(len(_p5) == 1, f"g18: exactly one pt5 artifact (got {_p5})")
_p5doc = json.loads(Path(_p5[0]).read_text())
check(_p5doc["point"] == "override_capture"
and _p5doc["calibration_signal_not_validated"] is True,
f"g18: pt5 carries point=override_capture + literal "
f"calibration_signal_not_validated:true (got "
f"{_p5doc.get('calibration_signal_not_validated')!r})")
check(_p5doc["actual"] is None and _p5doc["exit_error_summary"]
and _p5doc["predicted_vs_actual_delta_mib"] is None,
f"g18: boot never reached allocation -> actual null, "
f"exit_error_summary set, delta null (got actual="
f"{_p5doc['actual']!r}, exit={_p5doc['exit_error_summary']!r})")
check("predicted_b_breakdown" in _p5doc,
"g18: pt5 carries the full [B] kv-calc predicted breakdown")
for _x in r.capture_paths:
try:
os.unlink(_x)
except OSError:
pass
# --- g19: override-accepted WITHOUT --force-download -> NO [E] -----------
# Without --force-download the wont-fit advisory is NOT satisfied (the [C1]
# row needs --force-download); honest non-pass, no [E].
c = _Calls()
r = P.run_pull(DSLUG, "vllm/minimal", path="B", hardware_sm=SM_86,
fetcher=ff_derived(DSLUG, dense_cfg("Qwen2ForCausalLM"),
weight_gb=400.0),
profiles=profiles, statvfs=BIG_DISK, trust_remote_code=True,
gpu_topology=TOPO_2, **mk_emocks(c))
check(not r.ok and r.abort_reason.startswith("override-accepted")
and (c.emit + c.dl + c.boot + c.cap + c.p5) == 0,
f"g19: override-accepted WITHOUT --force-download -> NOT satisfied, "
f"NO [E] (ok={r.ok}, reason={r.abort_reason}, emit={c.emit})")
# --- g20: confirm→proceed WITHOUT --yes -> NOT eligible, NO [E] ----------
c = _Calls()
r = P.run_pull(DSLUG, "vllm/minimal", path="B", hardware_sm=SM_86,
fetcher=ff_derived(DSLUG, dense_cfg("Qwen2ForCausalLM"),
weight_gb=4.0),
profiles=profiles, statvfs=BIG_DISK, trust_remote_code=True,
gpu_topology=TOPO_2, **mk_emocks(c))
check(not r.ok and r.abort_reason.startswith("confirm→proceed")
and (c.emit + c.dl + c.boot + c.cap + c.p5) == 0,
f"g20: confirm→proceed WITHOUT --yes -> NOT download-eligible, "
f"NO [E] (ok={r.ok}, reason={r.abort_reason}, emit={c.emit})")
# --- g21: curated Path-A -> NO download stage (unchanged) ----------------
c = _Calls()
_o21 = "/tmp/_pull_g21.yml"
if os.path.exists(_o21):
os.unlink(_o21)
r = P.run_pull(CURATED_SLUG, "vllm/dual", path="A", out=_o21,
hardware_sm=SM_86, fetcher=NoNet(), profiles=profiles,
statvfs=BIG_DISK, yes=True, gpu_topology=TOPO_2,
**mk_emocks(c))
check(r.ok and r.emitted and r.path == "A"
and (c.emit + c.dl + c.boot + c.cap + c.p5) == 0,
f"g21: curated Path-A emits via [D], NO derived [E] download stage "
f"(emitted={r.emitted}, emit={c.emit} dl={c.dl})")
check(r.download_ok is None and r.boot_ok is None,
"g21: curated Path-A leaves [E] additive fields untouched (unchanged)")
if os.path.exists(_o21):
os.unlink(_o21)
# --- g22: derived but CONTRACT-5 reject -> structured refuse, NO dl/boot -
# Point --profile-like at an overlay/TQ3 shape so derived_emittable refuses
# BEFORE any download/boot. vllm/gemma-int8-tq3 = TQ3 KV + required feats.
c = _Calls()
r = P.run_pull(DSLUG, "vllm/gemma-int8-tq3", path="B", hardware_sm=SM_90,
fetcher=ff_derived(DSLUG, dense_cfg("Qwen2ForCausalLM"),
weight_gb=4.0),
profiles=profiles, statvfs=BIG_DISK, trust_remote_code=True,
yes=True, gpu_topology=TOPO_2, **mk_emocks(c))
_n22 = " ".join(r.notices)
check("derived-runtime-unsupported:" in _n22
and (c.emit + c.dl + c.boot + c.cap + c.p5) == 0,
f"g22: CONTRACT-5 reject -> structured derived-runtime-unsupported "
f"refuse, NO emit/download/boot (notices={_n22!r}, emit={c.emit} "
f"dl={c.dl})")
check(r.download_ok is None and r.boot_ok is None,
"g22: CONTRACT-5 reject leaves [E] additive fields None (no stage ran)")
# Purge ALL mocked-[E] capture residue (leave NO repo artifact).
_purge_captures()
# ===========================================================================
# Done.
# ===========================================================================
if failures:
print(f"\nSUMMARY: {len(failures)} assertion(s) FAILED.",
file=sys.stderr)
for f in failures:
print(f" - {f}", file=sys.stderr)
sys.exit(1)
print("\nSUMMARY: all Pull-Gate P4 truth-table assertions passed "
"(§4.1 9 cells + 6 strata + ordering + g0..g15 + trc-leak + "
"Path-B isolation) + v0.8.0 [E] E4 orchestration (g16..g22: "
"derived-[E] continuation, E3/E4-fix boot-lifecycle CM ordering "
"[g16b up->smoke->capture->down, smoke+capture see server ALIVE] + "
"[g16c teardown-on-exception], --dry-run verdict-only, "
"--force-download override + pt5 force-capture, confirm-without-yes/"
"override-without-force gating, curated-Path-A unchanged, "
"CONTRACT-5 reject).")
PY
echo "test-pull.sh OK"
Reference in New Issue View Git Blame Copy Permalink