diff --git a/CHANGELOG.md b/CHANGELOG.md index 2be49b1c..8a3352bb 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -21,6 +21,7 @@ Planned as 0.12.0. ### Added +- **`SIEVAL_TARGET_GPU` declares the GPUs a serving plan is resolved for**, as `::` with the name and memory `nvidia-smi` reports, e.g. `NVIDIA H100 80GB HBM3:8:81559`. Recipe resolution reads it instead of `nvidia-smi`, on any node, so a CPU node resolves the plan, and the inference identity, a node with those GPUs would. Unset, nothing changes. Declared, an engine launches only on a node with exactly those GPUs. `infer_plans.yaml` gains `gpu`, the GPUs its plans were resolved for, outside every identity; a `sieval run --resume` declaring the same GPUs reuses the recorded plan of each served model whose `path`, `infer`, capability layer and `deterministic` are unchanged, unless the plan held an engine secret, which `infer_plans.yaml` records redacted (#215, closes #212). - `sieval.core.runners.task_resume_plan` / `plan_resume`: the per-stage resume plan, without running anything. ## [0.11.0] - 2026-10-23 diff --git a/docs/guide/infer.md b/docs/guide/infer.md index 8a3821d6..1b754db1 100644 --- a/docs/guide/infer.md +++ b/docs/guide/infer.md @@ -53,6 +53,35 @@ The instruct flags are inert on a base checkpoint — the engine accepts and ignores them — so this affects what `infer_plans.yaml` records as the params used, not whether the service starts. +## Resolving for declared GPUs + +A recipe's hardware layer and its tensor and data parallelism come from the +GPUs `nvidia-smi` reports. To resolve a plan on another node, a CPU node +included, declare those GPUs instead: + +```bash +export SIEVAL_TARGET_GPU='NVIDIA H100 80GB HBM3:8:81559' # :: +``` + +`` and `` are what +`nvidia-smi --query-gpu=name,memory.total --format=csv,noheader,nounits` prints +on a node with those GPUs, and `` is how many it lists. The plan is then +the one that node resolves, so the inference identity matches too. A shorthand +like `H100-80G:8` is refused: the name is recorded in a path-only plan and the +memory decides tensor parallelism, so only the reported values reproduce it. + +Declared, `sieval infer start` and `sieval run` launch an engine only on a node +with exactly those GPUs, memory included: two nodes of one model can report +different `memory.total` (ECC on or off, PCIe or SXM), and the refusal names +each field that differs. `--dry-run` checks nothing, so it runs on any node. + +`infer_plans.yaml` records as `gpu` the GPUs its plans were resolved for, +declared or detected. A `sieval run --resume` declaring the same GPUs reuses a +recorded plan, rather than resolving it again, for each served model whose +`path`, `infer`, capability layer and `deterministic` are unchanged. A plan +holding an engine `api_key` or `authorization` is resolved again instead: +`infer_plans.yaml` writes those values as ``. + ## Handles and engine logs Each launched service has a handle file under `~/.sieval/handles`: diff --git a/sieval/cli/infer/commands.py b/sieval/cli/infer/commands.py index a62987d3..39935ef7 100644 --- a/sieval/cli/infer/commands.py +++ b/sieval/cli/infer/commands.py @@ -36,6 +36,7 @@ DeployError, DeployTimeoutError, ) +from sieval.infer.introspect import check_launch_gpu from sieval.infer.params import merge_params from sieval.infer.recipes import capability_model_type from sieval.infer.topology.models import ( @@ -376,6 +377,8 @@ async def _resolve() -> ResolveResult: render(dr_result, output) return + # A plan resolved for declared GPUs launches only on those GPUs. + anyio.run(check_launch_gpu) keep = resolve_keep_logs(keep_logs) # Claim the handle under the handle lock: a live holder refuses, and a diff --git a/sieval/cli/infer/recipe.py b/sieval/cli/infer/recipe.py index 0f8084fa..baf3988d 100644 --- a/sieval/cli/infer/recipe.py +++ b/sieval/cli/infer/recipe.py @@ -24,6 +24,7 @@ GPUInfo, ModelIdentity, bytes_per_param, + declared_target_gpu, detect_local_gpu, introspect_checkpoint_sync, ) @@ -274,7 +275,9 @@ async def _resolve_recipe_params( # final merge operate on a single canonical key form. overrides = merge_params(overrides or {}) - gpu = await detect_local_gpu() + # The declared GPUs stand in for this node's, so any node resolves the plan + # a node with that hardware would. + gpu = declared_target_gpu() or await detect_local_gpu() params: dict[str, ParamValue] = {} if gpu: diff --git a/sieval/cli/run.py b/sieval/cli/run.py index 9bf5ee12..e1a930c7 100644 --- a/sieval/cli/run.py +++ b/sieval/cli/run.py @@ -47,7 +47,13 @@ ) from sieval.cli.resolution import resolve_config_model_types from sieval.cli.services import SERVICE_LOG_DIR -from sieval.cli.session import resolve_deterministic, unwrap_proxies +from sieval.cli.session import ( + holds_secret, + read_persisted_config, + redact_secrets, + resolve_deterministic, + unwrap_proxies, +) from sieval.cli.signals import terminate_as_interrupt from sieval.core.models import Deployment from sieval.core.utils.logging import configure_logging, log_user @@ -55,8 +61,14 @@ from sieval.infer.backends.translator import BackendCommand, inject_user_env from sieval.infer.config import InferHandle from sieval.infer.deployer import DEFAULT_READY_TIMEOUT, LocalDeployer +from sieval.infer.introspect import ( + check_launch_gpu, + declared_target_gpu, + detect_local_gpu, + format_target_gpu, +) from sieval.infer.recipes import capability_model_type -from sieval.infer.topology import DeploymentPlan +from sieval.infer.topology import DeploymentPlan, plan_from_mapping from sieval.infer.topology.resolver import auto_resolve_plan @@ -94,16 +106,122 @@ def _refuse_explicit_endpoint(model_name: str, model_config: dict) -> None: ) +def _recorded_session_dir( + resume: bool | str | Path | None, result_dir: str | None +) -> Path | None: + """The session ``result_dir`` a resume continues, whose plans it may reuse. + + ``--resume DIR`` names it, or names a task directory inside it. + """ + if resume is True: + return Path(result_dir) if result_dir else None + if isinstance(resume, (str, Path)): + recorded = Path(resume) + return recorded.parent if (recorded / "meta.json").is_file() else recorded + return None + + +class _RecordedPlans: + """What a resumed session recorded about its served models, read-only. + + Its plans are reused only when ``SIEVAL_TARGET_GPU`` declares the GPUs they + were resolved for, and only for a model whose serving config is unchanged: + a recipe the installed SiEval resolves differently would otherwise change + the inference identity and refuse the resume. + """ + + def __init__(self, session_dir: Path | None, gpu: str | None) -> None: + self._plans: dict[str, Any] = {} + self._models: dict[str, Any] = {} + self._capabilities: dict[str, str] = {} + if session_dir is None or gpu is None: + return + recorded = read_persisted_config(session_dir / "infer_plans.yaml") + if recorded is None or recorded.get("gpu") != gpu: + return + plans = recorded.get("models") + config = read_persisted_config(session_dir / "effective_config.yaml") + if not isinstance(plans, dict) or config is None: + return + models = config.get("models") + try: + types = resolve_config_model_types(config).model_types_by_config + except ValueError: + return + if isinstance(models, dict): + self._plans, self._models = plans, models + self._capabilities = { + name: capability_model_type(kind) for name, kind in types.items() + } + + def plan_for( + self, + model_name: str, + model_config: dict[str, Any], + capability: str, + deterministic: bool, + ) -> DeploymentPlan | None: + """The recorded plan of *model_name*, when this run would serve it alike.""" + raw = self._plans.get(model_name) + recorded_model = self._models.get(model_name) + if not isinstance(raw, dict) or not isinstance(recorded_model, dict): + return None + # `infer_plans.yaml` is written redacted: a plan that held a secret + # (an engine `api_key`) reads back holding the marker, and would + # launch the engine with it. + if holds_secret(raw): + return None + # Compared redacted, as `effective_config.yaml` records them. + current = redact_secrets(model_config) + if any(recorded_model.get(k) != current.get(k) for k in ("path", "infer")): + return None + if self._capabilities.get(model_name) != capability: + return None + try: + plan = plan_from_mapping(raw) + except ValueError: + return None + if plan.deterministic != deterministic: + return None + return plan + + +def _declared_gpu_text() -> str | None: + """``SIEVAL_TARGET_GPU``, normalized; ``None`` when nothing is declared.""" + declared = declared_target_gpu() + return format_target_gpu(declared) if declared is not None else None + + +async def _plan_gpu_text() -> str | None: + """The GPUs this run's plans were resolved for, as ``SIEVAL_TARGET_GPU`` + spells them: the declared ones, else this node's.""" + gpu = declared_target_gpu() or await detect_local_gpu() + return format_target_gpu(gpu) if gpu is not None else None + + +def _recorded_plans( + resume: bool | str | Path | None, result_dir: str | None +) -> _RecordedPlans: + return _RecordedPlans( + _recorded_session_dir(resume, result_dir), _declared_gpu_text() + ) + + async def _prepare_launch_batch( config_path: Path, config: dict[str, Any], *, effective_deterministic: bool, + recorded: _RecordedPlans | None = None, ) -> tuple[ dict[str, dict[str, Any]], dict[str, tuple[DeploymentPlan, list[BackendCommand]]], ]: - """Resolve and translate every managed deployment without launching it.""" + """Resolve and translate every managed deployment without launching it. + + A plan *recorded* by the session a resume continues is reused when it + applies (:class:`_RecordedPlans`). + """ models = config.get("models", {}) if not isinstance(models, dict): @@ -120,7 +238,23 @@ async def _prepare_launch_batch( _refuse_explicit_endpoint(model_name, model_config) infer_dict = model_config.get("infer") - if infer_dict is not None: + capability = capability_model_type( + model_type_resolution.model_types_by_config[model_name] + ) + reused = ( + recorded.plan_for( + model_name, model_config, capability, effective_deterministic + ) + if recorded is not None + else None + ) + if reused is not None: + logger.info("Reusing the recorded serving plan of {}", model_name) + plan = reused + user_env = { + str(k): str(v) for k, v in ((infer_dict or {}).get("env") or {}).items() + } + elif infer_dict is not None: _, plan, user_env = await resolve_infer_config( config_path, model_name, @@ -129,10 +263,7 @@ async def _prepare_launch_batch( else: checkpoint = model_config["path"] result = await auto_resolve_plan( - checkpoint=checkpoint, - capability=capability_model_type( - model_type_resolution.model_types_by_config[model_name] - ), + checkpoint=checkpoint, capability=capability ) plan = result.plan user_env = {} @@ -208,6 +339,7 @@ def checks( config_path, loaded, effective_deterministic=effective_deterministic, + recorded=_recorded_plans(resume, result_dir or loaded.get("result_dir")), ) return checks(plan_dicts or None, frozenset(prepared_launches)) @@ -272,6 +404,7 @@ async def _run_all( config_path, config, effective_deterministic=effective_deterministic, + recorded=_recorded_plans(resume, effective_result_dir), ) # Task requirements, model declarations, and every managed desired @@ -314,6 +447,8 @@ async def _run_all( f"Model name {SERVICE_LOG_DIR!r} is reserved for " f"/logs/{SERVICE_LOG_DIR}/; rename the model." ) + if prepared_launches: + await check_launch_gpu() for model_name, (plan, commands) in prepared_launches.items(): log_user("Starting inference for model: {}", model_name) @@ -406,6 +541,7 @@ def _progress( invocation=invocation, realized_deployments=realized_deployments or None, cli_model=cli_model, + infer_gpu=await _plan_gpu_text() if plan_dicts else None, ) return reports diff --git a/sieval/cli/session.py b/sieval/cli/session.py index 5dd2fd5b..6bb8b21a 100644 --- a/sieval/cli/session.py +++ b/sieval/cli/session.py @@ -1089,7 +1089,7 @@ def _strip_header(text: str) -> str: _REDACTED = "" -def _redact_secrets(value: Any) -> Any: +def redact_secrets(value: Any) -> Any: """A copy of *value* with every secret config value, at any depth, redacted. Matched by key name anywhere, so a task arg named ``api_key`` is redacted @@ -1099,14 +1099,30 @@ def _redact_secrets(value: Any) -> Any: return { k: _REDACTED if k in _SECRET_CONFIG_KEYS and v is not None - else _redact_secrets(v) + else redact_secrets(v) for k, v in value.items() } if isinstance(value, list): - return [_redact_secrets(item) for item in value] + return [redact_secrets(item) for item in value] return value +def holds_secret(value: Any) -> bool: + """Whether :func:`redact_secrets` redacts anything in *value*. + + A record written through :func:`redact_secrets` holds the marker in its + place, so such a record cannot stand in for the value it was written from. + """ + if isinstance(value, Mapping): + return any( + (k in _SECRET_CONFIG_KEYS and v is not None) or holds_secret(v) + for k, v in value.items() + ) + if isinstance(value, list): + return any(holds_secret(item) for item in value) + return False + + def _diff_lines(a: dict[str, Any], b: dict[str, Any]) -> list[str]: """Return ``- : → `` lines for every differing leaf. @@ -1286,7 +1302,7 @@ def _fold_runner_resume_aliases(cfg: dict[str, Any]) -> None: block["resume"] = str(resume) if isinstance(resume, Path) else resume -def _read_persisted_config(path: Path) -> dict[str, Any] | None: +def read_persisted_config(path: Path) -> dict[str, Any] | None: try: loaded = yaml.safe_load(path.read_text(encoding="utf-8")) except (OSError, yaml.YAMLError): @@ -1654,6 +1670,7 @@ def __init__( serving_reconciler: ServingReconciler | None = None, resume_from: str | Path | None = None, cli_model: CliModel | None = None, + infer_gpu: str | None = None, ): self.config_path = Path(config_path) self.model_override = model_override @@ -1677,6 +1694,7 @@ def __init__( ) self._legacy_endpoint_map: Mapping[str, str] = legacy_endpoint_map self._infer_plans: Mapping[str, dict[str, Any]] | None = infer_plans + self._infer_gpu = infer_gpu self._realized_deployments = realized_deployment_map for root_name, deployment in self._realized_deployments.items(): if not isinstance(root_name, str) or not root_name: @@ -4893,7 +4911,7 @@ def _legacy_match_for( ``None`` when there is none: both stages stay unknown. """ if isinstance(runner_config.resume, Path): - persisted = _read_persisted_config( + persisted = read_persisted_config( runner_config.resume.parent / "effective_config.yaml" ) elif runner_config.resume: @@ -5161,7 +5179,7 @@ async def _persist_effective_config(self) -> None: ``path`` / ``infer`` fields. """ body = yaml.safe_dump( - _redact_secrets(self._reified_config), + redact_secrets(self._reified_config), default_flow_style=False, allow_unicode=True, sort_keys=False, @@ -5169,7 +5187,7 @@ async def _persist_effective_config(self) -> None: result_dir = self._resolve_result_dir() if self.resume_override is True and result_dir: self._recorded_config = await run_sync( - _read_persisted_config, Path(result_dir) / "effective_config.yaml" + read_persisted_config, Path(result_dir) / "effective_config.yaml" ) extra_lines = [ "Reproduce:", @@ -5203,16 +5221,21 @@ async def _persist_infer_plans(self) -> None: """Write infer_plans.yaml to result_dir when the caller supplied plans. Rerun re-resolves plans from the ``infer:`` section of - effective_config.yaml plus the installed sieval version, so this file - is audit-only. It never blocks a resume: the serving recipe is part of - each task's inference identity, which a resume compares instead. + effective_config.yaml plus the installed sieval version. A resume + under ``SIEVAL_TARGET_GPU`` equal to the recorded ``gpu`` reuses a plan + whose model config is unchanged and that holds no redacted secret + (``sieval run``). It never blocks a resume: the serving recipe is part + of each task's inference identity, which a resume compares instead. """ if not self._infer_plans: return - payload = {"models": dict(self._infer_plans)} + payload: dict[str, Any] = {"models": dict(self._infer_plans)} + if self._infer_gpu is not None: + # Outside `models`, so it stays out of every inference identity. + payload = {"gpu": self._infer_gpu, **payload} body = yaml.safe_dump( - _redact_secrets(payload), + redact_secrets(payload), default_flow_style=False, allow_unicode=True, sort_keys=False, @@ -5224,7 +5247,10 @@ async def _persist_infer_plans(self) -> None: extra_lines=[ "Reference only: audit log of the DeploymentPlan used", " for each served model. Re-resolved at runtime from", - " effective_config.yaml + installed sieval version.", + " effective_config.yaml + installed sieval version, except", + " that a resume reuses a model's plan when SIEVAL_TARGET_GPU", + " equals `gpu`, the model's config is unchanged and its", + " plan holds no value.", ], ) await self._persist_audit_yaml( @@ -5304,6 +5330,7 @@ async def arun_session( realized_deployments: Mapping[str, Deployment] | None = None, resume_from: str | Path | None = None, cli_model: CliModel | None = None, + infer_gpu: str | None = None, ) -> dict[str, Any]: """Run tasks defined in a YAML configuration file asynchronously. @@ -5333,6 +5360,9 @@ async def arun_session( resume_from: Deprecated alias of ``resume=``; warns. cli_model: ``--checkpoint``/``--api-base``, added as the ``models:`` entry of a config that has none. + infer_gpu: The GPUs *infer_plans* were resolved for, as + ``SIEVAL_TARGET_GPU`` spells them; recorded in ``infer_plans.yaml`` + as ``gpu``, so a resume can tell whether it may reuse the plans. Returns: A dictionary mapping task names to their reports. @@ -5350,6 +5380,7 @@ async def arun_session( realized_deployments=realized_deployments, resume_from=resume_from, cli_model=cli_model, + infer_gpu=infer_gpu, ) return await runner.arun() diff --git a/sieval/infer/introspect.py b/sieval/infer/introspect.py index d04d27b2..91312a8e 100644 --- a/sieval/infer/introspect.py +++ b/sieval/infer/introspect.py @@ -13,8 +13,10 @@ AI-Generated Code - Claude Opus 4.6 (Anthropic) """ +import os import re import subprocess +from collections.abc import Mapping from dataclasses import dataclass from pathlib import Path from typing import Any @@ -477,3 +479,79 @@ async def detect_local_gpu() -> GPUInfo | None: info.memory_mib, ) return info + + +#: The GPUs a serving plan is resolved for, read instead of ``nvidia-smi``. +TARGET_GPU_ENV = "SIEVAL_TARGET_GPU" + +_TARGET_GPU_FORM = ( + "'::', with and as " + "`nvidia-smi --query-gpu=name,memory.total --format=csv,noheader,nounits` " + "prints them on that node, e.g. 'NVIDIA H100 80GB HBM3:8:81559'" +) + + +def parse_target_gpu(text: str) -> GPUInfo: + """Parse a GPU declaration, ``::``. + + The fields are the ones :func:`detect_local_gpu` reads, verbatim: the name + selects the recipe's hardware profile and is recorded in a path-only plan, + and the memory decides tensor parallelism. A shorthand such as + ``H100-80G`` would resolve another plan than the node does. + """ + parts = text.rsplit(":", 2) + name = parts[0].strip() if len(parts) == 3 else "" + try: + count, memory_mib = int(parts[1]), int(parts[2]) + except (IndexError, ValueError): + count = memory_mib = 0 + if not name or count < 1 or memory_mib < 1: + raise ValueError(f"{TARGET_GPU_ENV}={text!r} is not {_TARGET_GPU_FORM}") + return GPUInfo(model=name, count=count, memory_mib=memory_mib) + + +def format_target_gpu(gpu: GPUInfo) -> str: + """The declaration :func:`parse_target_gpu` reads back as *gpu*.""" + return f"{gpu.model}:{gpu.count}:{gpu.memory_mib}" + + +def declared_target_gpu(environ: Mapping[str, str] | None = None) -> GPUInfo | None: + """The GPUs :data:`TARGET_GPU_ENV` declares; ``None`` when it is unset.""" + raw = (os.environ if environ is None else environ).get(TARGET_GPU_ENV, "") + return parse_target_gpu(raw) if raw.strip() else None + + +async def check_launch_gpu() -> None: + """Refuse to launch an engine on GPUs other than the declared ones. + + A plan resolved for :data:`TARGET_GPU_ENV` is only that node's plan on the + declared hardware, so every field must match, memory included: two nodes + of one SKU can report different ``memory.total`` (ECC, PCIe vs SXM), and + the memory decides tensor parallelism. Without a declaration nothing is + checked. + """ + declared = declared_target_gpu() + if declared is None: + return + detected = await detect_local_gpu() + if detected == declared: + return + if detected is None: + here = "no GPU" + else: + differing = [ + f"{field} {theirs!r} (declared {ours!r})" + for field, ours, theirs in ( + ("name", declared.model, detected.model), + ("count", declared.count, detected.count), + ("memory MiB", declared.memory_mib, detected.memory_mib), + ) + if ours != theirs + ] + here = f"{format_target_gpu(detected)!r}: {', '.join(differing)}" + raise RuntimeError( + f"The serving plan was resolved for {TARGET_GPU_ENV}=" + f"{format_target_gpu(declared)!r}, but this node has {here}. " + "Declare the GPUs this node has, or run on a node that has the " + "declared ones." + ) diff --git a/sieval/infer/topology/__init__.py b/sieval/infer/topology/__init__.py index a4e2fbf3..1bbb2d54 100644 --- a/sieval/infer/topology/__init__.py +++ b/sieval/infer/topology/__init__.py @@ -26,6 +26,7 @@ WellKnownRole, deployment_plan_fingerprint, deployment_plan_projection, + plan_from_mapping, ) from sieval.infer.topology.resolver import precision_key from sieval.infer.topology.validator import validate_plan @@ -52,6 +53,7 @@ "WellKnownRole", "deployment_plan_fingerprint", "deployment_plan_projection", + "plan_from_mapping", "precision_key", "validate_plan", ] diff --git a/sieval/infer/topology/models.py b/sieval/infer/topology/models.py index 04dbd063..42e2b6ec 100644 --- a/sieval/infer/topology/models.py +++ b/sieval/infer/topology/models.py @@ -265,6 +265,44 @@ def deployment_plan_projection( ) +def plan_from_mapping(raw: Mapping[str, Any]) -> DeploymentPlan: + """Rebuild a :class:`DeploymentPlan` from its plain form, as + ``infer_plans.yaml`` records it. + + Raises ``ValueError`` for a mapping that is not such a plan. + """ + try: + assignments = tuple( + RoleAssignment( + role=a["role"], + devices=DeviceGroup(**a["devices"]), + topology=ParallelTopology(**a["topology"]), + replicas=a.get("replicas", 1), + engine_params=dict(a.get("engine_params") or {}), + scaling=( + ScalingPolicy(**a["scaling"]) + if a.get("scaling") is not None + else None + ), + ) + for a in raw["assignments"] + ) + attachments = tuple(ServiceBinding(**b) for b in raw.get("attachments") or ()) + plan = DeploymentPlan( + checkpoint=raw["checkpoint"], + backend=raw["backend"], + assignments=assignments, + attachments=attachments, + deterministic=raw.get("deterministic", False), + seed=raw.get("seed", DETERMINISTIC_DEFAULT_SEED), + ) + except (KeyError, TypeError) as exc: + raise ValueError(f"not a recorded deployment plan: {exc}") from exc + if _plain_plan_value(plan) != _plain_plan_value(raw): + raise ValueError("recorded deployment plan has fields this version lacks") + return plan + + def _plain_plan_value(value: Any) -> Any: """Recursively normalize plan dataclasses and frozen mappings for JSON.""" diff --git a/sieval/infer/topology/resolver.py b/sieval/infer/topology/resolver.py index c8e617e6..0f5f3c40 100644 --- a/sieval/infer/topology/resolver.py +++ b/sieval/infer/topology/resolver.py @@ -13,8 +13,10 @@ from sieval.infer.config import ParamValue from sieval.infer.introspect import ( + TARGET_GPU_ENV, ModelIdentity, bytes_per_param, + declared_target_gpu, detect_local_gpu, extract_moe_info, introspect_checkpoint_with_config, @@ -410,10 +412,13 @@ async def auto_resolve_plan( family=identity.family, ) - # 2. Detect GPU - gpu = await detect_local_gpu() + # 2. The declared GPUs, else this node's + gpu = declared_target_gpu() or await detect_local_gpu() if gpu is None: - raise RuntimeError("No GPU detected — cannot auto-resolve deployment plan.") + raise RuntimeError( + "No GPU detected — cannot auto-resolve deployment plan. On a node " + f"without the GPUs, declare them in {TARGET_GPU_ENV}." + ) hw = HardwareEnv( gpu_count=gpu.count, diff --git a/tests/conftest.py b/tests/conftest.py index 273dcd9c..db10ac3a 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -270,6 +270,19 @@ def _neutralize_openai_client_finalizer(): yield +# =================================================================== +# No declared GPUs leak in from the shell (autouse) +# =================================================================== +@pytest.fixture(autouse=True) +def _no_declared_target_gpu(monkeypatch: pytest.MonkeyPatch) -> None: + """Unset ``SIEVAL_TARGET_GPU``, which replaces GPU detection. + + A test that fakes ``detect_local_gpu`` would otherwise resolve for whatever + the developer's shell declares. + """ + monkeypatch.delenv("SIEVAL_TARGET_GPU", raising=False) + + # =================================================================== # Mock Dataset # =================================================================== diff --git a/tests/unit/cli/test_run.py b/tests/unit/cli/test_run.py index 604ae938..416df60b 100644 --- a/tests/unit/cli/test_run.py +++ b/tests/unit/cli/test_run.py @@ -15,6 +15,7 @@ from typer.testing import CliRunner from sieval.cli import app +from sieval.cli.run import _RecordedPlans from sieval.core.models import ( Deployment, DeploymentTopology, @@ -1516,3 +1517,260 @@ async def stub_prepare(self): loaded = yaml.safe_load(infer_plans.read_text()) assert "model_a" in loaded["models"] assert loaded["models"]["model_a"]["backend"] == "vllm" + + +class TestDeclaredTargetGpu: + """``SIEVAL_TARGET_GPU`` resolves on any node the plan a node with those + GPUs resolves, and a resume reuses the plan it recorded for them.""" + + H100 = "NVIDIA H100 80GB HBM3:8:81559" + + def _config(self, tmp_path: Path) -> Path: + checkpoint = tmp_path / "Qwen3-32B" + checkpoint.mkdir() + (checkpoint / "config.json").write_text( + json.dumps( + { + "architectures": ["Qwen3ForCausalLM"], + "model_type": "qwen3", + "vocab_size": 151936, + "hidden_size": 5120, + "num_hidden_layers": 64, + "num_attention_heads": 64, + "num_key_value_heads": 8, + "intermediate_size": 25600, + "torch_dtype": "bfloat16", + } + ) + ) + config_path = tmp_path / "cfg.yaml" + config_path.write_text( + yaml.safe_dump( + { + "models": {"m": {"path": str(checkpoint), "type": "chat"}}, + "result_dir": str(tmp_path / "out"), + "tasks": {}, + } + ) + ) + return config_path + + async def _run( + self, config_path: Path, *, gpu: object, resume: bool = False + ) -> tuple[AsyncMock, AsyncMock]: + """Run with this node's GPUs faked as *gpu*; return the launch and + session mocks.""" + from sieval.cli.run import _run_all + + launch = AsyncMock(return_value=([_fake_handle()], None)) + session = AsyncMock(return_value={}) + nvidia_smi = AsyncMock(return_value=gpu) + with ( + patch("sieval.infer.topology.resolver.detect_local_gpu", new=nvidia_smi), + patch("sieval.infer.introspect.detect_local_gpu", new=nvidia_smi), + patch("sieval.cli.run.detect_local_gpu", new=nvidia_smi), + patch("sieval.cli.run.launch_model", new=launch), + patch("sieval.cli.run.cleanup_model", new=AsyncMock()), + patch("sieval.cli.session.arun_session", new=session), + ): + await _run_all(config_path, resume=resume) + return launch, session + + @staticmethod + def _gpu(text: str) -> object: + from sieval.infer.introspect import parse_target_gpu + + return parse_target_gpu(text) + + @pytest.mark.anyio + async def test_a_declared_plan_is_the_detected_one( + self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch + ): + config_path = self._config(tmp_path) + _, detected = await self._run(config_path, gpu=self._gpu(self.H100)) + + monkeypatch.setenv("SIEVAL_TARGET_GPU", self.H100) + _, declared = await self._run(config_path, gpu=self._gpu(self.H100)) + + assert declared.await_args is not None and detected.await_args is not None + plans = declared.await_args.kwargs["infer_plans"] + assert plans == detected.await_args.kwargs["infer_plans"] + # Both name the GPUs the plans are for, as the variable spells them. + assert declared.await_args.kwargs["infer_gpu"] == self.H100 + assert detected.await_args.kwargs["infer_gpu"] == self.H100 + + @pytest.mark.anyio + async def test_declared_gpus_other_than_this_nodes_refuse_the_launch( + self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch + ): + monkeypatch.setenv("SIEVAL_TARGET_GPU", self.H100) + with pytest.raises(RuntimeError, match="no GPU"): + await self._run(self._config(tmp_path), gpu=None) + + def _record(self, tmp_path: Path, config_path: Path, plan: dict) -> None: + """What a session that served ``m`` with *plan* on the H100s left.""" + out = tmp_path / "out" + out.mkdir(exist_ok=True) + (out / "effective_config.yaml").write_text(config_path.read_text()) + (out / "infer_plans.yaml").write_text( + yaml.safe_dump({"gpu": self.H100, "models": {"m": plan}}, sort_keys=False) + ) + + @pytest.mark.anyio + async def test_a_resume_reuses_the_plan_recorded_for_the_declared_gpus( + self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch + ): + from sieval.cli.session import unwrap_proxies + + config_path = self._config(tmp_path) + # Not what this SiEval resolves, so only a reuse produces it. + recorded = unwrap_proxies(_fake_plan()) + recorded["checkpoint"] = yaml.safe_load(config_path.read_text())["models"]["m"][ + "path" + ] + self._record(tmp_path, config_path, recorded) + monkeypatch.setenv("SIEVAL_TARGET_GPU", self.H100) + resolve = AsyncMock() + with patch("sieval.cli.run.auto_resolve_plan", new=resolve): + _, session = await self._run( + config_path, gpu=self._gpu(self.H100), resume=True + ) + + resolve.assert_not_awaited() + assert session.await_args is not None + assert session.await_args.kwargs["infer_plans"] == {"m": recorded} + + @pytest.mark.anyio + async def test_a_plan_recorded_with_a_secret_is_resolved_again( + self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch + ): + """``infer_plans.yaml`` is written redacted, so reusing a plan with an + engine ``api_key`` would launch the engine with the marker as its key.""" + from sieval.cli.session import redact_secrets, unwrap_proxies + + config_path = self._config(tmp_path) + recorded = unwrap_proxies(_fake_plan()) + recorded["checkpoint"] = yaml.safe_load(config_path.read_text())["models"]["m"][ + "path" + ] + recorded["assignments"][0]["engine_params"]["api_key"] = "sk-engine" + self._record(tmp_path, config_path, redact_secrets(recorded)) + monkeypatch.setenv("SIEVAL_TARGET_GPU", self.H100) + resolve = AsyncMock(return_value=ResolveResult(plan=_fake_plan(), steps=())) + with patch("sieval.cli.run.auto_resolve_plan", new=resolve): + launch, _ = await self._run( + config_path, gpu=self._gpu(self.H100), resume=True + ) + + resolve.assert_awaited_once() + assert launch.await_count == 1 + assert "" not in str(launch.await_args_list) + + def _recorded_plans( + self, tmp_path: Path, plan: dict | None = None + ) -> _RecordedPlans: + from sieval.cli.session import unwrap_proxies + + config_path = self._config(tmp_path) + raw = plan if plan is not None else unwrap_proxies(_fake_plan()) + self._record(tmp_path, config_path, raw) + return _RecordedPlans(tmp_path / "out", self.H100) + + def _model_config(self, tmp_path: Path) -> dict: + return yaml.safe_load((tmp_path / "cfg.yaml").read_text())["models"]["m"] + + def test_a_recorded_plan_applies_only_to_the_same_serving(self, tmp_path: Path): + recorded = self._recorded_plans(tmp_path) + model = self._model_config(tmp_path) + plan_for = recorded.plan_for + + assert plan_for("m", model, "instruct", False) == _fake_plan() + assert plan_for("other", model, "instruct", False) is None + assert plan_for("m", {**model, "path": "/elsewhere"}, "instruct", False) is None + assert plan_for("m", model, "base", False) is None + assert plan_for("m", model, "instruct", True) is None + + def test_a_record_this_version_cannot_read_is_not_reused(self, tmp_path: Path): + from sieval.cli.session import unwrap_proxies + + raw = unwrap_proxies(_fake_plan()) + raw["added_later"] = 1 + recorded = self._recorded_plans(tmp_path, raw) + model = self._model_config(tmp_path) + assert recorded.plan_for("m", model, "instruct", False) is None + + @pytest.mark.parametrize( + "damage", ["no_plans_file", "no_models", "no_config", "bad_config"] + ) + def test_an_incomplete_record_reuses_nothing(self, tmp_path: Path, damage: str): + self._recorded_plans(tmp_path) + out = tmp_path / "out" + if damage == "no_plans_file": + (out / "infer_plans.yaml").unlink() + if damage == "no_models": + (out / "infer_plans.yaml").write_text(yaml.safe_dump({"gpu": self.H100})) + if damage == "no_config": + (out / "effective_config.yaml").unlink() + if damage == "bad_config": + # A model the session's type resolution refuses. + (out / "effective_config.yaml").write_text( + yaml.safe_dump({"models": {"m": "not a mapping"}, "tasks": {}}) + ) + recorded = _RecordedPlans(out, self.H100) + model = self._model_config(tmp_path) + assert recorded.plan_for("m", model, "instruct", False) is None + + @pytest.mark.parametrize( + ("resume", "expected"), + [ + (True, "out"), + (False, None), + (None, None), + ("session", "session"), + ("session/task", "session"), + ], + ) + def test_the_recorded_session_dir( + self, tmp_path: Path, resume: bool | str | None, expected: str | None + ): + from sieval.cli.run import _recorded_session_dir + + (tmp_path / "session" / "task").mkdir(parents=True) + (tmp_path / "session" / "task" / "meta.json").write_text("{}") + target = tmp_path / resume if isinstance(resume, str) else resume + found = _recorded_session_dir(target, str(tmp_path / "out")) + assert found == (tmp_path / expected if expected else None) + # A plain resume without a result_dir has nothing to read. + assert _recorded_session_dir(True, None) is None + + @pytest.mark.anyio + @pytest.mark.parametrize("change", ["other_gpus", "fresh_run", "changed_model"]) + async def test_otherwise_the_plan_is_resolved_again( + self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch, change: str + ): + from sieval.cli.session import unwrap_proxies + + config_path = self._config(tmp_path) + self._record(tmp_path, config_path, unwrap_proxies(_fake_plan())) + declared = self.H100 + if change == "other_gpus": + declared = "NVIDIA A100-SXM4-80GB:8:81920" + if change == "changed_model": + cfg = yaml.safe_load(config_path.read_text()) + cfg["models"]["m"]["infer"] = {"backend": "vllm"} + cfg["models"]["m"]["infer"]["overrides"] = {"tp": 8} + config_path.write_text(yaml.safe_dump(cfg)) + monkeypatch.setenv("SIEVAL_TARGET_GPU", declared) + resolve = AsyncMock(return_value=ResolveResult(plan=_fake_plan(), steps=())) + resolve_yaml = AsyncMock(return_value=("m", _fake_plan(), {})) + with ( + patch("sieval.cli.run.auto_resolve_plan", new=resolve), + patch("sieval.cli.run.resolve_infer_config", new=resolve_yaml), + ): + await self._run( + config_path, + gpu=self._gpu(declared), + resume=change != "fresh_run", + ) + + assert resolve.await_count + resolve_yaml.await_count == 1 diff --git a/tests/unit/cli/test_session.py b/tests/unit/cli/test_session.py index 1ea71a31..3efdcab4 100644 --- a/tests/unit/cli/test_session.py +++ b/tests/unit/cli/test_session.py @@ -7439,6 +7439,7 @@ async def test_arun_session_delegates_to_runner(self): realized_deployments=None, resume_from=None, cli_model=None, + infer_gpu=None, ) assert result == {"task_a": {"ok": True}} @@ -7472,6 +7473,7 @@ async def test_arun_session_passes_deterministic(self): realized_deployments=None, resume_from=None, cli_model=None, + infer_gpu=None, ) assert result == {"task_a": {"ok": True}} @@ -7534,6 +7536,7 @@ async def test_arun_session_forwards_legacy_external_endpoint_adapter(self): realized_deployments=None, resume_from=None, cli_model=None, + infer_gpu=None, ) def test_run_session_forwards_legacy_external_endpoint_adapter_positionally( @@ -9699,6 +9702,33 @@ def test_secret_values_are_redacted_but_variable_names_are_not(self): ] +class TestHoldsSecret: + @pytest.mark.parametrize( + "value", + [ + {"api_key": "k"}, + {"a": [{"b": {"authorization": "Bearer x"}}]}, + ], + ) + def test_whatever_redact_secrets_would_redact(self, value): + from sieval.cli.session import holds_secret, redact_secrets + + assert redact_secrets(value) != value + assert holds_secret(value) + # And so does what it wrote: the marker stands where the secret was. + assert holds_secret(redact_secrets(value)) + + @pytest.mark.parametrize( + "value", + [{"api_key": None}, {"api_key_env": "KEY"}, ["api_key"], "api_key", {}], + ) + def test_nothing_else(self, value): + from sieval.cli.session import holds_secret, redact_secrets + + assert not holds_secret(value) + assert redact_secrets(value) == value + + # ── Tests for env expansion + error-hint wrapping in _setup_datasets ── diff --git a/tests/unit/infer/test_target_gpu.py b/tests/unit/infer/test_target_gpu.py new file mode 100644 index 00000000..7bcd9f5c --- /dev/null +++ b/tests/unit/infer/test_target_gpu.py @@ -0,0 +1,223 @@ +"""Tests for ``SIEVAL_TARGET_GPU``: declared GPUs in place of ``nvidia-smi``. + +A plan resolved from a declaration must be the plan a node with those GPUs +resolves, byte for byte, since the serving recipe is part of the inference +identity a resume compares. + +AI-Generated Code - Claude Opus 5.5 (Anthropic) +""" + +import json +from pathlib import Path +from unittest.mock import AsyncMock, patch + +import pytest +import yaml + +from sieval.cli.infer.recipe import resolve_infer_config +from sieval.cli.session import unwrap_proxies +from sieval.infer.introspect import ( + TARGET_GPU_ENV, + GPUInfo, + check_launch_gpu, + declared_target_gpu, + format_target_gpu, + parse_target_gpu, +) +from sieval.infer.topology import plan_from_mapping +from sieval.infer.topology.resolver import auto_resolve_plan + +# As `nvidia-smi --query-gpu=name,memory.total --format=csv,noheader,nounits` +# prints an 8-GPU H100 node. +H100 = GPUInfo(model="NVIDIA H100 80GB HBM3", count=8, memory_mib=81559) +A100_40G = GPUInfo(model="NVIDIA A100-SXM4-40GB", count=4, memory_mib=40960) + +_RECIPE_GPU = "sieval.cli.infer.recipe.detect_local_gpu" +_RESOLVER_GPU = "sieval.infer.topology.resolver.detect_local_gpu" +_LAUNCH_GPU = "sieval.infer.introspect.detect_local_gpu" + + +def _qwen3_checkpoint(root: Path, name: str = "Qwen3-32B") -> Path: + checkpoint = root / name + checkpoint.mkdir() + (checkpoint / "config.json").write_text( + json.dumps( + { + "architectures": ["Qwen3ForCausalLM"], + "model_type": "qwen3", + "vocab_size": 151936, + "hidden_size": 5120, + "num_hidden_layers": 64, + "num_attention_heads": 64, + "num_key_value_heads": 8, + "intermediate_size": 25600, + "torch_dtype": "bfloat16", + } + ) + ) + return checkpoint + + +class TestParseTargetGpu: + def test_reads_the_fields_nvidia_smi_reports(self) -> None: + assert parse_target_gpu("NVIDIA H100 80GB HBM3:8:81559") == H100 + + def test_a_name_may_hold_colons(self) -> None: + gpu = parse_target_gpu("vendor:card:2:1024") + assert gpu == GPUInfo(model="vendor:card", count=2, memory_mib=1024) + + def test_round_trips_through_its_text(self) -> None: + assert parse_target_gpu(format_target_gpu(H100)) == H100 + + @pytest.mark.parametrize( + "text", + ["H100-80G:8", "H100", ":8:81559", "H100:0:81559", "H100:8:0", "H100:x:1"], + ) + def test_refuses_anything_else_naming_the_form(self, text: str) -> None: + with pytest.raises(ValueError, match=r"::"): + parse_target_gpu(text) + + def test_unset_or_blank_declares_nothing(self) -> None: + assert declared_target_gpu({}) is None + assert declared_target_gpu({TARGET_GPU_ENV: " "}) is None + assert declared_target_gpu({TARGET_GPU_ENV: format_target_gpu(H100)}) == H100 + + +class TestDeclaredEqualsDetected: + """The plan from a declaration is the one detection on that node gives.""" + + @pytest.mark.anyio + @pytest.mark.parametrize("gpu", [H100, A100_40G], ids=["h100x8", "a100-40gx4"]) + async def test_path_only_plan( + self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch, gpu: GPUInfo + ) -> None: + checkpoint = str(_qwen3_checkpoint(tmp_path)) + with patch(_RESOLVER_GPU, new=AsyncMock(return_value=gpu)): + detected = await auto_resolve_plan(checkpoint, capability="instruct") + + monkeypatch.setenv(TARGET_GPU_ENV, format_target_gpu(gpu)) + nvidia_smi = AsyncMock(return_value=None) # a CPU node + with patch(_RESOLVER_GPU, new=nvidia_smi): + declared = await auto_resolve_plan(checkpoint, capability="instruct") + + nvidia_smi.assert_not_awaited() + assert declared.plan == detected.plan + assert yaml.safe_dump(unwrap_proxies(declared.plan), sort_keys=False) == ( + yaml.safe_dump(unwrap_proxies(detected.plan), sort_keys=False) + ) + + @pytest.mark.anyio + @pytest.mark.parametrize("gpu", [H100, A100_40G], ids=["h100x8", "a100-40gx4"]) + async def test_yaml_infer_plan( + self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch, gpu: GPUInfo + ) -> None: + checkpoint = str(_qwen3_checkpoint(tmp_path)) + config = tmp_path / "cfg.yaml" + config.write_text( + yaml.safe_dump( + { + "models": { + "m": {"infer": {"backend": "vllm", "checkpoint": checkpoint}} + }, + "tasks": {}, + } + ) + ) + with patch(_RECIPE_GPU, new=AsyncMock(return_value=gpu)): + detected = await resolve_infer_config(config, "m") + + monkeypatch.setenv(TARGET_GPU_ENV, format_target_gpu(gpu)) + nvidia_smi = AsyncMock(return_value=None) + with patch(_RECIPE_GPU, new=nvidia_smi): + declared = await resolve_infer_config(config, "m") + + nvidia_smi.assert_not_awaited() + assert declared.plan == detected.plan + # The hardware layer applied: an A100-40G and an H100 differ. + assert declared.plan.assignments[0].engine_params + + @pytest.mark.anyio + async def test_another_declaration_resolves_another_plan( + self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch + ) -> None: + """The check above has teeth: the declaration reaches the plan.""" + checkpoint = str(_qwen3_checkpoint(tmp_path)) + plans = [] + for gpu in (H100, A100_40G): + monkeypatch.setenv(TARGET_GPU_ENV, format_target_gpu(gpu)) + plans.append((await auto_resolve_plan(checkpoint)).plan) + assert plans[0] != plans[1] + + @pytest.mark.anyio + async def test_no_gpu_and_no_declaration_names_the_variable( + self, tmp_path: Path + ) -> None: + checkpoint = str(_qwen3_checkpoint(tmp_path)) + with ( + patch(_RESOLVER_GPU, new=AsyncMock(return_value=None)), + pytest.raises(RuntimeError, match=TARGET_GPU_ENV), + ): + await auto_resolve_plan(checkpoint) + + +class TestCheckLaunchGpu: + @pytest.mark.anyio + async def test_nothing_declared_checks_nothing(self) -> None: + nvidia_smi = AsyncMock(return_value=None) + with patch(_LAUNCH_GPU, new=nvidia_smi): + await check_launch_gpu() + nvidia_smi.assert_not_awaited() + + @pytest.mark.anyio + async def test_the_declared_gpus_launch( + self, monkeypatch: pytest.MonkeyPatch + ) -> None: + monkeypatch.setenv(TARGET_GPU_ENV, format_target_gpu(H100)) + with patch(_LAUNCH_GPU, new=AsyncMock(return_value=H100)): + await check_launch_gpu() + + @pytest.mark.anyio + @pytest.mark.parametrize( + ("here", "named"), + [(A100_40G, "NVIDIA A100-SXM4-40GB:4:40960"), (None, "no GPU")], + ) + async def test_other_gpus_are_refused( + self, monkeypatch: pytest.MonkeyPatch, here: GPUInfo | None, named: str + ) -> None: + monkeypatch.setenv(TARGET_GPU_ENV, format_target_gpu(H100)) + with ( + patch(_LAUNCH_GPU, new=AsyncMock(return_value=here)), + pytest.raises(RuntimeError, match=named), + ): + await check_launch_gpu() + + @pytest.mark.anyio + async def test_the_refusal_names_the_field_that_differs( + self, monkeypatch: pytest.MonkeyPatch + ) -> None: + # One SKU, another reported memory.total (e.g. ECC on). + here = GPUInfo(model=H100.model, count=H100.count, memory_mib=80000) + monkeypatch.setenv(TARGET_GPU_ENV, format_target_gpu(H100)) + with ( + patch(_LAUNCH_GPU, new=AsyncMock(return_value=here)), + pytest.raises(RuntimeError) as refused, + ): + await check_launch_gpu() + message = str(refused.value) + assert f"memory MiB 80000 (declared {H100.memory_mib})" in message + assert "name " not in message and "count " not in message + + +class TestPlanFromMapping: + @pytest.mark.anyio + async def test_a_recorded_plan_reads_back_equal( + self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch + ) -> None: + monkeypatch.setenv(TARGET_GPU_ENV, format_target_gpu(H100)) + plan = (await auto_resolve_plan(str(_qwen3_checkpoint(tmp_path)))).plan + recorded = yaml.safe_load(yaml.safe_dump(unwrap_proxies(plan), sort_keys=False)) + assert plan_from_mapping(recorded) == plan + + def test_an_unknown_field_is_refused(self) -> None: + with pytest.raises(ValueError): + plan_from_mapping({"checkpoint": "c", "backend": "vllm"})