From 18d5d20b4ad29ccd9bfff46a836d4f57c8901923 Mon Sep 17 00:00:00 2001 From: Chien-Chun Hung <2679986+chienchunhung@users.noreply.github.com> Date: Thu, 1 Oct 2026 15:17:33 -0700 Subject: [PATCH 1/4] fix: resolve and validate disaggregated precheck cache contracts Signed-off-by: Chien-Chun Hung <2679986+chienchunhung@users.noreply.github.com> --- .../cache_transceiver_precheck/README.md | 30 ++- .../precheck_config.py | 43 ++++- .../run_precheck.py | 139 ++++++++++---- ...dep8_gen1_dep8_eplb0_dspark3_ccb-NIXL.yaml | 1 + ...ep8_gen1_dep16_eplb0_dspark5_ccb-NIXL.yaml | 1 + ...dep8_gen1_dep8_eplb0_dspark3_ccb-NIXL.yaml | 1 + ...ep8_gen1_dep16_eplb0_dspark5_ccb-NIXL.yaml | 1 + .../test_cache_transceiver_precheck_e2e.py | 36 ++++ .../test_cache_transceiver_precheck_config.py | 92 +++++++++ .../test_cache_transceiver_precheck_run.py | 179 ++++++++++++++++++ 10 files changed, 489 insertions(+), 34 deletions(-) diff --git a/tests/scripts/perf-sanity/cache_transceiver_precheck/README.md b/tests/scripts/perf-sanity/cache_transceiver_precheck/README.md index 49f8e19c3597..0e4aa14d991b 100644 --- a/tests/scripts/perf-sanity/cache_transceiver_precheck/README.md +++ b/tests/scripts/perf-sanity/cache_transceiver_precheck/README.md @@ -16,7 +16,7 @@ starts. | Same UCX env vars (incl. the `unset UCX_TLS` cases) | `jenkins/scripts/perf/submit.py` builds the precheck commands from the **same** `ucx_tls_cmd` + `$CTX/GEN_WORKER_ENV_VARS` strings as the worker steps; `slurm_precheck_run.sh` sources the same `slurm_env_setup.sh` (the `UCX_TLS=tcp` fixup) as `slurm_run.sh`. | | Same instance count / parallelism | One precheck `srun` per ctx/gen server with the same `-N/--ntasks/--ntasks-per-node/--mpi=pmix` and the same node slices (`-w`) as the real server steps (`slurm_launch_draft.sh`). TP/PP/CP/attention-DP come from the same `worker_config`. | | Same transceiver config | `CacheTransceiverConfig(**yaml["worker_config"][role]["cache_transceiver_config"])` — the yaml block is passed through verbatim (backend, `max_tokens_in_buffer`, timeouts, ...). | -| Same KV cache manager version + transceiver runtime | The launch generator forwards the real test's `LLM_MODELS_ROOT` into every precheck process. Explicit per-side `kv_cache_config.use_kv_cache_manager_v2` wins; absent means "auto" and requires a registered model class, then resolves via `get_preferred_kv_cache_manager_version()`. `transceiver_runtime: auto` resolves via `get_preferred_transceiver_runtime()` (NIXL-gated) — both through the same llm_utils code serving uses. An unresolved manager-version `auto` setting fails with INIT_ERROR instead of silently assuming V1. V2 requires the Python transceiver (the C++ one only supports V1); a V2+CPP combination also fails fast. | +| Same KV cache manager version + transceiver runtime | The launch generator forwards the real test's `LLM_MODELS_ROOT` into every precheck process. Explicit per-side `kv_cache_config.use_kv_cache_manager_v2` wins; absent means "auto" and requires a registered model class, then resolves via `get_preferred_kv_cache_manager_version()`. `transceiver_runtime: auto` resolves via `get_preferred_transceiver_runtime()` (NIXL-gated) — both through the same llm_utils code serving uses. The model class comes from the checkpoint; when no checkpoint directory is resolved, `metadata.architectures` supplies a registered architecture hint. An unresolved manager-version `auto` setting fails with INIT_ERROR instead of silently assuming V1. V2 requires the Python transceiver (the C++ one only supports V1); a V2+CPP combination also fails fast. | Asymmetric layouts (ctx dep4 → gen dep16, ctx pp8 → gen tp32, ...) are supported: data is seeded per (request, **global** layer) and constant along @@ -24,6 +24,34 @@ the KV-head axis, so the receiver regenerates its expected slice locally under any TP/PP resharding. KV shape (layers/heads/head_dim, MLA vs GQA, MTP nextn layers) is read from the real model's `config.json` under `$LLM_MODELS_ROOT`. +For checkpoints unavailable to the network precheck, declare the model architecture: + +```yaml +metadata: + architectures: [DeepseekV4ForCausalLM] +``` + +The first declaration is looked up in the model registry. A resolved checkpoint +always takes precedence; metadata does not replace an unknown or invalid +checkpoint architecture. Model hooks still run through the serving resolvers, +and a hook that needs unavailable checkpoint fields fails closed. + +Without a checkpoint, the precheck transfers a generic synthetic KV pool. +A PASS establishes transport availability, not checkpoint readiness or exact +model KV-layout correctness. Serving still requires the model weights and +`config.json` staged under `LLM_MODELS_ROOT` on every worker. + +Before creating the KV manager or transceiver, all ranks in an instance must +agree on the resolved settings for both roles and global synthetic KV shape. +The context/generation handshake compares this same contract before any +transfer: effective backend, runtime, manager version, normalized dtype, +tokens per block, and global layers/heads/head dimension/MLA mode. Comparing +both role resolutions preserves configured asymmetric manager and parallelism +choices while rejecting inconsistent checkpoint visibility or environment +resolution. Local checkpoint paths and diagnostic source labels are excluded. +Architecture declarations, manager settings, and dtype inputs also participate +in the shared plan fingerprint. + ## Enabling / disabling **On by default** for every disaggregated perf-sanity test. To opt out: diff --git a/tests/scripts/perf-sanity/cache_transceiver_precheck/precheck_config.py b/tests/scripts/perf-sanity/cache_transceiver_precheck/precheck_config.py index 497edae4377d..ccc45396eff4 100644 --- a/tests/scripts/perf-sanity/cache_transceiver_precheck/precheck_config.py +++ b/tests/scripts/perf-sanity/cache_transceiver_precheck/precheck_config.py @@ -88,7 +88,7 @@ # Fallback KV shape for dry-runs and explicitly selected manager versions when # the model directory cannot be resolved. Manager-version "auto" resolution -# fails fast instead of silently pairing this shape with V1. +# requires a registered checkpoint or declared architecture instead of assuming V1. FALLBACK_KV_SHAPE = { "num_layers": 32, "num_kv_heads": 8, @@ -339,6 +339,40 @@ def gate_library_content(draft_launch_sh, llm_src): ) +def declared_architectures(cfg: Mapping[str, Any]) -> list[str]: + """Read architecture hints for prechecks without a local checkpoint. + + Args: + cfg: Parsed disaggregated benchmark YAML. + + Returns: + Architecture names in declared priority order, or an empty list. + + Raises: + ValueError: If the declaration is not a string or list of nonempty strings. + """ + value = (cfg.get("metadata") or {}).get("architectures", []) + if isinstance(value, str): + value = [value] + if not isinstance(value, list) or any( + not isinstance(name, str) or not name.strip() for name in value + ): + raise ValueError("metadata.architectures must be a string or list of nonempty strings") + return [name.strip() for name in value] + + +def normalized_kv_dtype(dtype: str) -> str: + """Return the storage dtype used by the synthetic KV pool. + + Args: + dtype: Per-role KV dtype from the benchmark configuration. + + Returns: + Canonical fp8, fp16, or bf16 spelling; unsupported/auto uses bf16. + """ + return {"fp8": "fp8", "fp16": "fp16", "half": "fp16"}.get(dtype.lower(), "bf16") + + def resolve_plan(cfg, benchmark_mode="e2e"): """Build the shared precheck plan both roles must agree on. @@ -439,6 +473,7 @@ def resolve_plan(cfg, benchmark_mode="e2e"): ), "rendezvous_timeout_s": rendezvous_timeout_s, "verify_data": bool(knobs["verify_data"]), + "architectures": declared_architectures(cfg), } for role, side, xcvr in (("ctx", ctx_side, ctx_xcvr), ("gen", gen_side, gen_xcvr)): kv_cfg = side.get("kv_cache_config") or {} @@ -457,6 +492,11 @@ def resolve_plan(cfg, benchmark_mode="e2e"): def plan_fingerprint(plan): """Stable string both sides must agree on before transferring.""" keys = ( + "architectures", + "ctx_use_kv_cache_manager_v2", + "gen_use_kv_cache_manager_v2", + "ctx_kv_dtype", + "gen_kv_dtype", "num_ctx_servers", "num_gen_servers", "ctx", @@ -479,6 +519,7 @@ def side_plan(plan, role): """Per-role view: this role's parallelism + transceiver/kv config.""" return { "role": role, + "architectures": plan["architectures"], "parallel": plan[role], "cache_transceiver_config": plan[f"{role}_cache_transceiver_config"], "kv_dtype": plan[f"{role}_kv_dtype"], diff --git a/tests/scripts/perf-sanity/cache_transceiver_precheck/run_precheck.py b/tests/scripts/perf-sanity/cache_transceiver_precheck/run_precheck.py index 0519cb5dfce2..b72756319c35 100644 --- a/tests/scripts/perf-sanity/cache_transceiver_precheck/run_precheck.py +++ b/tests/scripts/perf-sanity/cache_transceiver_precheck/run_precheck.py @@ -241,6 +241,7 @@ def load_internal_apis(): ) from tensorrt_llm._torch.distributed import Distributed from tensorrt_llm._torch.models.modeling_utils import get_registered_model_class + from tensorrt_llm._torch.pyexecutor.config_utils import resolve_cache_transceiver_config from tensorrt_llm._torch.pyexecutor.hang_detector import HangDetector from tensorrt_llm._torch.pyexecutor.kv_cache.kv_cache_manager_v2 import KVCacheManagerV2 from tensorrt_llm._torch.pyexecutor.llm_request import ( @@ -288,6 +289,7 @@ def load_internal_apis(): TorchLlmArgs=TorchLlmArgs, resolve_kv_cache_manager_v2_auto=_resolve_kv_cache_manager_v2_auto, resolve_transceiver_runtime_auto=_resolve_transceiver_runtime_auto, + resolve_cache_transceiver_config=resolve_cache_transceiver_config, Mapping=Mapping, SamplingParams=SamplingParams, ) @@ -390,11 +392,16 @@ class and adopt its get_preferred_kv_cache_manager_version() value api = load_internal_apis() model_cls, hf_view = _lookup_model_cls(model_dir) + declared = side.get("architectures", []) + if model_dir is None and declared: + model_cls = api.get_registered_model_class(declared[0]) setting = side["use_kv_cache_manager_v2"] if setting == "auto" and model_cls is None: raise RuntimeError( "use_kv_cache_manager_v2 is 'auto', but the precheck could not resolve " - f"a registered model class from model_dir={model_dir!r}; refusing to assume V1" + f"a registered model class from model_dir={model_dir!r} or " + f"metadata.architectures={declared!r}; refusing to assume V1. " + "Check LLM_MODELS_ROOT/checkpoint staging or declare the model architecture." ) # Runtime BEFORE V2, like serving: the V2 resolver's disagg gating reads @@ -413,7 +420,7 @@ class and adopt its get_preferred_kv_cache_manager_version() value try: parallel = side["parallel"] llm_args_kwargs = { - "model": model_dir, + "model": model_dir or "", "tensor_parallel_size": parallel["tp"], "pipeline_parallel_size": parallel["pp"], "context_parallel_size": parallel["cp"], @@ -445,8 +452,8 @@ def build_kv_cache_manager(kv_shape, plan, side, mapping, max_req_len, use_v2): "bf16": api.DataType.BF16, } dtype_str = side["kv_dtype"].lower() - dtype = dtype_map.get(dtype_str) - if dtype is None: + dtype = dtype_map[pcfg.normalized_kv_dtype(dtype_str)] + if dtype_str not in dtype_map: print(f"[precheck] kv dtype {dtype_str!r} not mapped, using BF16", flush=True) dtype = api.DataType.BF16 @@ -929,6 +936,7 @@ def __init__(self, args, plan, side, comm): self.runtime = "CPP" # Resolved in setup(): "auto" needs the model preference hook. self.use_v2 = False + self.transfer_contract = {} self.mapping = None self.llm_request_state = None self.csv_dir = os.path.join(self.work_dir, "csv", f"{self.role}_{self.server_idx}") @@ -944,9 +952,74 @@ def _consensus_error(self, local_err): return f"rank(s) {ranks}: {bad[0][1]}" # ---- setup ------------------------------------------------------------- - def setup(self, kv_shape, max_req_len): + def _resolve_transfer_config(self, kv_shape: dict) -> object: + """Resolve local settings without constructing GPU or distributed resources. + + Args: + kv_shape: Global synthetic cache dimensions resolved on this rank. + + Returns: + The resolved CacheTransceiverConfig consumed by setup. + + Raises: + RuntimeError: If preferences cannot be resolved or V2 requires Python. + """ api = load_internal_apis() + role_configs = {} + contracts = {} + # Both role resolutions are compared, preserving supported asymmetry + # (e.g. a V1 sender and V2 receiver) while detecting local model drift. + for role in ("ctx", "gen"): + side = pcfg.side_plan(self.plan, role) + cache_cfg = api.CacheTransceiverConfig(**side["cache_transceiver_config"]) + use_v2 = resolve_model_prefs(self.plan.get("_model_dir"), side, cache_cfg) + api.resolve_cache_transceiver_config(cache_cfg) + runtime = cache_cfg.transceiver_runtime or "CPP" + if use_v2 and runtime != "PYTHON": + raise RuntimeError( + "KVCacheManagerV2 requires cache_transceiver_config." + f"transceiver_runtime: PYTHON, got {runtime!r}" + ) + role_configs[role] = cache_cfg + contracts[role] = { + "use_v2": use_v2, + "runtime": runtime, + "backend": cache_cfg._resolve_default_backend()[0], + "kv_dtype": pcfg.normalized_kv_dtype(side["kv_dtype"]), + } + self.use_v2 = contracts[self.role]["use_v2"] + self.transfer_contract = { + **contracts, + "tokens_per_block": self.plan["tokens_per_block"], + "kv_shape": { + key: kv_shape[key] for key in ("num_layers", "num_kv_heads", "head_dim", "is_mla") + }, + } + if self.is_leader and (kv_shape.get("simplified") or self.plan.get("_model_dir") is None): + print( + f"[precheck {self.role}_{self.server_idx}] synthetic KV pool; " + f"model_dir={self.plan.get('_model_dir')!r}, " + f"architectures={self.plan['architectures']!r}. " + "This checks cache transport; serving still requires a staged checkpoint.", + flush=True, + ) + return role_configs[self.role] + def setup(self, kv_shape, max_req_len): + cache_cfg = None + local_err = None + try: + cache_cfg = self._resolve_transfer_config(kv_shape) + except Exception as exc: # noqa: BLE001 - every rank must reach consensus + local_err = exc + reason = self._consensus_error(local_err) + if reason is not None: + raise RuntimeError(f"model/cache configuration resolution failed: {reason}") + contracts = self.comm.allgather(self.transfer_contract) + if any(contract != contracts[0] for contract in contracts[1:]): + raise RuntimeError(f"resolved transfer contract mismatch within instance: {contracts}") + + api = load_internal_apis() self.llm_request_state = api.LlmRequestState par = self.side["parallel"] self.mapping = api.Mapping( @@ -966,30 +1039,6 @@ def setup(self, kv_shape, max_req_len): # side). os.environ["TRTLLM_KVCACHE_TIME_OUTPUT_PATH"] = self.csv_dir - # Built VERBATIM from the disagg yaml's cache_transceiver_config so - # backend/max_tokens_in_buffer/timeouts match the real test exactly. - cache_cfg = api.CacheTransceiverConfig(**self.side["cache_transceiver_config"]) - # Yaml-absent settings resolve against the model's preferences, like - # serving does (kv manager version + transceiver runtime) -- this holds - # even for the simplified stand-in pool: only the KV SHAPE is generic; - # the V1/V2 manager version and the transceiver runtime must still - # match what the real model runs (e.g. V4 -> V2 + Python). - self.use_v2 = resolve_model_prefs(self.plan.get("_model_dir"), self.side, cache_cfg) - if kv_shape.get("simplified") and self.is_leader: - print( - f"[precheck {self.role}_{self.server_idx}] SIMPLIFIED: {kv_shape['simplified']}", - flush=True, - ) - # KVCacheManagerV2 only works with the Python transceiver (see - # cache_transceiver_test/report.py); reject the pairing up front with - # a clear INIT_ERROR instead of a C++ binding type error. - if self.use_v2 and cache_cfg.transceiver_runtime != "PYTHON": - raise RuntimeError( - "KVCacheManagerV2 requires cache_transceiver_config." - f"transceiver_runtime: PYTHON, got {cache_cfg.transceiver_runtime!r} " - "(the C++ transceiver only supports the V1 manager)" - ) - manager_cls = api.KVCacheManagerV2 if self.use_v2 else api.KVCacheManager api.maybe_enable_fabric_memory_for_python_transceiver(cache_cfg, manager_cls) self.kvm = build_kv_cache_manager( @@ -1480,7 +1529,7 @@ def _recv_ctx_control( # test, never over ZMQ): # # gen leader ctx leader -# | -- hello {fingerprint} --------------> | yaml mismatch -> abort +# | -- hello {fingerprint, contract} ----> | config mismatch -> abort # | <---------------- welcome ------------ | # | -- go {li, rep, wave} --------------> | every rank posts its sends # | <----- params {pair: ctx_phase} ------ | (from the owning dp ranks) @@ -1515,7 +1564,19 @@ def leader_reply(obj): if msg[0] != "hello" or msg[1].get("fingerprint") != plan["fingerprint"]: leader_reply(("abort", "plan fingerprint mismatch (ctx/gen yaml disagree)")) raise _TransferError(f"handshake with gen_{peer_idx} failed: {msg[:1]}") - leader_reply(("welcome", {"fingerprint": plan["fingerprint"]})) + if msg[1].get("transfer_contract") != runner.transfer_contract: + reason = ( + "resolved transfer contract mismatch: " + f"ctx={runner.transfer_contract}, gen={msg[1].get('transfer_contract')}" + ) + leader_reply(("abort", reason)) + raise _TransferError(reason) + leader_reply( + ( + "welcome", + {"fingerprint": plan["fingerprint"], "transfer_contract": runner.transfer_contract}, + ) + ) for li, req_len, rep, wave in _schedule(plan): msg = _recv_ctx_control( @@ -1634,7 +1695,14 @@ def _gen_open_session(runner, peer_idx, arm): try: reply = runner._leader_send_recv_with_progress( sock, - ("hello", {"gen_idx": runner.server_idx, "fingerprint": plan["fingerprint"]}), + ( + "hello", + { + "gen_idx": runner.server_idx, + "fingerprint": plan["fingerprint"], + "transfer_contract": runner.transfer_contract, + }, + ), key, peer_role="ctx", peer_idx=peer_idx, @@ -1646,6 +1714,13 @@ def _gen_open_session(runner, peer_idx, arm): raise _TransferError(f"ctx_{peer_idx} aborted handshake: {reply[1]}") if reply[0] != "welcome": raise _TransferError(f"unexpected handshake reply from ctx_{peer_idx}: {reply[:1]}") + if ( + reply[1].get("fingerprint") != plan["fingerprint"] + or reply[1].get("transfer_contract") != runner.transfer_contract + ): + reason = f"ctx_{peer_idx} handshake contract mismatch: {reply[1]}" + runner._leader_send_recv(sock, ("abort", reason), key) + raise _TransferError(reason) if runner.is_leader: zmq, _ = runner._zmq() sock.setsockopt(zmq.RCVTIMEO, hello_s * 1000) diff --git a/tests/scripts/perf-sanity/disaggregated/gb300_deepseek-v4-pro-dspark_agentx_con1156_ctx2_dep8_gen1_dep8_eplb0_dspark3_ccb-NIXL.yaml b/tests/scripts/perf-sanity/disaggregated/gb300_deepseek-v4-pro-dspark_agentx_con1156_ctx2_dep8_gen1_dep8_eplb0_dspark3_ccb-NIXL.yaml index 11991961ec4b..8ee1d9bf96d4 100644 --- a/tests/scripts/perf-sanity/disaggregated/gb300_deepseek-v4-pro-dspark_agentx_con1156_ctx2_dep8_gen1_dep8_eplb0_dspark3_ccb-NIXL.yaml +++ b/tests/scripts/perf-sanity/disaggregated/gb300_deepseek-v4-pro-dspark_agentx_con1156_ctx2_dep8_gen1_dep8_eplb0_dspark3_ccb-NIXL.yaml @@ -1,4 +1,5 @@ metadata: + architectures: [DeepseekV4ForCausalLM] model_name: deepseek_v4_pro_nvfp4_dspark precision: fp4 model_dir_name: DeepSeek-V4-Pro-nvfp4-DSpark diff --git a/tests/scripts/perf-sanity/disaggregated/gb300_deepseek-v4-pro-dspark_agentx_con1456_ctx3_dep8_gen1_dep16_eplb0_dspark5_ccb-NIXL.yaml b/tests/scripts/perf-sanity/disaggregated/gb300_deepseek-v4-pro-dspark_agentx_con1456_ctx3_dep8_gen1_dep16_eplb0_dspark5_ccb-NIXL.yaml index e64a086320f4..b8c756cb4937 100644 --- a/tests/scripts/perf-sanity/disaggregated/gb300_deepseek-v4-pro-dspark_agentx_con1456_ctx3_dep8_gen1_dep16_eplb0_dspark5_ccb-NIXL.yaml +++ b/tests/scripts/perf-sanity/disaggregated/gb300_deepseek-v4-pro-dspark_agentx_con1456_ctx3_dep8_gen1_dep16_eplb0_dspark5_ccb-NIXL.yaml @@ -1,4 +1,5 @@ metadata: + architectures: [DeepseekV4ForCausalLM] model_name: deepseek_v4_pro_nvfp4_dspark precision: fp4 model_dir_name: DeepSeek-V4-Pro-nvfp4-DSpark diff --git a/tests/scripts/perf/disaggregated/gb300_deepseek-v4-pro-dspark_agentx_con1156_ctx2_dep8_gen1_dep8_eplb0_dspark3_ccb-NIXL.yaml b/tests/scripts/perf/disaggregated/gb300_deepseek-v4-pro-dspark_agentx_con1156_ctx2_dep8_gen1_dep8_eplb0_dspark3_ccb-NIXL.yaml index 11991961ec4b..8ee1d9bf96d4 100644 --- a/tests/scripts/perf/disaggregated/gb300_deepseek-v4-pro-dspark_agentx_con1156_ctx2_dep8_gen1_dep8_eplb0_dspark3_ccb-NIXL.yaml +++ b/tests/scripts/perf/disaggregated/gb300_deepseek-v4-pro-dspark_agentx_con1156_ctx2_dep8_gen1_dep8_eplb0_dspark3_ccb-NIXL.yaml @@ -1,4 +1,5 @@ metadata: + architectures: [DeepseekV4ForCausalLM] model_name: deepseek_v4_pro_nvfp4_dspark precision: fp4 model_dir_name: DeepSeek-V4-Pro-nvfp4-DSpark diff --git a/tests/scripts/perf/disaggregated/gb300_deepseek-v4-pro-dspark_agentx_con1456_ctx3_dep8_gen1_dep16_eplb0_dspark5_ccb-NIXL.yaml b/tests/scripts/perf/disaggregated/gb300_deepseek-v4-pro-dspark_agentx_con1456_ctx3_dep8_gen1_dep16_eplb0_dspark5_ccb-NIXL.yaml index e64a086320f4..b8c756cb4937 100644 --- a/tests/scripts/perf/disaggregated/gb300_deepseek-v4-pro-dspark_agentx_con1456_ctx3_dep8_gen1_dep16_eplb0_dspark5_ccb-NIXL.yaml +++ b/tests/scripts/perf/disaggregated/gb300_deepseek-v4-pro-dspark_agentx_con1456_ctx3_dep8_gen1_dep16_eplb0_dspark5_ccb-NIXL.yaml @@ -1,4 +1,5 @@ metadata: + architectures: [DeepseekV4ForCausalLM] model_name: deepseek_v4_pro_nvfp4_dspark precision: fp4 model_dir_name: DeepSeek-V4-Pro-nvfp4-DSpark diff --git a/tests/unittest/disaggregated/test_cache_transceiver_precheck_e2e.py b/tests/unittest/disaggregated/test_cache_transceiver_precheck_e2e.py index 9b106a1d8d43..cc6aa0b2d796 100644 --- a/tests/unittest/disaggregated/test_cache_transceiver_precheck_e2e.py +++ b/tests/unittest/disaggregated/test_cache_transceiver_precheck_e2e.py @@ -34,6 +34,7 @@ import subprocess import sys import time +from pathlib import Path import pytest import yaml @@ -354,3 +355,38 @@ def test_precheck_fails_fast_on_fingerprint_mismatch(tmp_path): # The gen driver saw the ctx abort its handshake: the reason must name it. text, _ = _read_status(work_dir, "gen_0") assert "fingerprint" in text or "abort" in text.lower() + + +@pytest.mark.timeout(300) +def test_precheck_declared_architecture_without_checkpoint(tmp_path: Path) -> None: + """Missing weights still allow actual V2/Python transfers after auto resolution.""" + pytest.importorskip("mpi4py") + cfg = _disagg_yaml(1, 1, 1, 1) + cfg["metadata"] = {"model_dir_name": "unstaged", "architectures": ["DeepseekV4ForCausalLM"]} + for side in cfg["worker_config"].values(): + side["kv_cache_config"].pop("use_kv_cache_manager_v2") + side["cache_transceiver_config"]["transceiver_runtime"] = "auto" + config_path, models_root = _write_inputs(tmp_path, cfg) + work_dir, launched = _launch_instances(tmp_path, _jobs(cfg, config_path), models_root) + _wait_all(launched) + _assert_all_passed(work_dir, launched) + + +@pytest.mark.timeout(300) +def test_precheck_rejects_inconsistent_checkpoint_visibility(tmp_path: Path) -> None: + """Peers with the same YAML plan reject different resolved synthetic layouts.""" + pytest.importorskip("mpi4py") + ctx_cfg = _disagg_yaml(1, 1, 1, 1) + ctx_path, models_root = _write_inputs(tmp_path, ctx_cfg, name="ctx") + gen_cfg = _disagg_yaml(1, 1, 1, 1) + gen_cfg["metadata"]["model_dir_name"] = "unstaged" + gen_path, _ = _write_inputs(tmp_path, gen_cfg, name="gen") + jobs = [("ctx", 0, 1, ctx_path), ("gen", 0, 1, gen_path)] + work_dir, launched = _launch_instances(tmp_path, jobs, models_root) + _wait_all(launched) + for name, proc, log_path in launched: + assert proc.returncode != 0, _log_tail(log_path) + text, doc = _read_status(work_dir, name) + assert "resolved transfer contract mismatch" in text + assert doc["overall"] == "FAIL" + assert not any(case["status"] == "PASS" for case in doc["cases"]) diff --git a/tests/unittest/others/test_cache_transceiver_precheck_config.py b/tests/unittest/others/test_cache_transceiver_precheck_config.py index 561d3107a05a..80e2b7c17764 100644 --- a/tests/unittest/others/test_cache_transceiver_precheck_config.py +++ b/tests/unittest/others/test_cache_transceiver_precheck_config.py @@ -965,6 +965,8 @@ def _run( num_ctx_servers=2, first_ctx_wave_delay_s=0, peer_progress_timeout_s=None, + contract_difference=None, + corrupt_welcome=False, ): import threading @@ -1008,6 +1010,23 @@ def _run( for i in range(num_ctx_servers) ] + if corrupt_welcome: + send_recv = gen._leader_send_recv_with_progress + + def change_welcome(*args: object, **kwargs: object) -> list: + """Corrupt only the welcome to exercise the gen-side guard.""" + reply = send_recv(*args, **kwargs) + if reply[0] == "welcome": + reply[1]["transfer_contract"] = {"unexpected": True} + return reply + + gen._leader_send_recv_with_progress = change_welcome + + if contract_difference is not None: + gen.transfer_contract = {"ctx": {"use_v2": True, "runtime": "PYTHON"}} + for ctx in ctxs: + ctx.transfer_contract = contract_difference + if fail_peer_idx is not None: real_gen_run_peer = rp.gen_run_peer @@ -1060,6 +1079,39 @@ def gen_arm(what, publish_progress=True, **kwargs): assert not leaked, f"ctx serve threads wedged: {leaked}" return plan, gen, ctxs, failures + @pytest.mark.parametrize( + "contract", + [ + {"ctx": {"use_v2": False, "runtime": "PYTHON"}}, + {"ctx": {"use_v2": True, "runtime": "CPP"}}, + {"ctx": {"use_v2": True, "runtime": "PYTHON"}, "kv_shape": {"is_mla": True}}, + ], + ids=["manager", "runtime", "shape"], + ) + def test_resolved_contract_mismatch_stops_before_transfer( + self, tmp_path: object, monkeypatch: pytest.MonkeyPatch, contract: dict + ) -> None: + """Real peer sessions reject divergent resolution without dispatching KV.""" + _, gen, ctxs, failures = self._run( + tmp_path, monkeypatch, num_ctx_servers=1, contract_difference=contract + ) + assert failures + assert all(ctx._calls["waves"] == 0 for ctx in ctxs) + assert any( + "resolved transfer contract mismatch" in case["reason"] for case in gen.recorder.cases + ) + + def test_rejected_welcome_aborts_peer( + self, tmp_path: object, monkeypatch: pytest.MonkeyPatch + ) -> None: + """Gen rejection completes the abort exchange so ctx does not time out.""" + _, gen, ctxs, failures = self._run( + tmp_path, monkeypatch, num_ctx_servers=1, corrupt_welcome=True + ) + assert failures == [("gen_0", "_PeerAbort")] + assert ctxs[0]._calls["waves"] == 0 + assert any("handshake contract mismatch" in case["reason"] for case in gen.recorder.cases) + def test_two_ctx_full_pass(self, tmp_path, monkeypatch): plan, gen, ctxs, failures = self._run(tmp_path, monkeypatch) assert not failures @@ -1390,3 +1442,43 @@ def test_python_transceiver_bandwidth_csv(tmp_path): assert abs(bw - 153600 * 1024 * 1024 / 1e9) < 1e-9 # no perf files -> None assert rp.parse_python_bandwidth_gbps(str(tmp_path / "empty")) is None + + +@pytest.mark.parametrize( + "value,expected", + [ + ([], []), + (" DeepseekV4ForCausalLM ", ["DeepseekV4ForCausalLM"]), + (["DeepseekV4ForCausalLM"], ["DeepseekV4ForCausalLM"]), + ], +) +def test_declared_architectures(value: object, expected: list[str]) -> None: + """Normalize metadata and retain it in the shared and per-role plans.""" + cfg = _disagg_yaml(metadata={"architectures": value}) + plan = pcfg.resolve_plan(cfg) + assert plan["architectures"] == expected + assert pcfg.side_plan(plan, "ctx")["architectures"] == expected + assert json.loads(plan["fingerprint"])["architectures"] == expected + + +@pytest.mark.parametrize("value", [None, 42, {}, [""], [" "], [None], [42]]) +def test_invalid_declared_architectures(value: object) -> None: + """Malformed hints fail early rather than selecting an unintended class.""" + with pytest.raises(ValueError, match="metadata.architectures"): + pcfg.resolve_plan(_disagg_yaml(metadata={"architectures": value})) + + +@pytest.mark.parametrize( + "field,value", + [ + ("architectures", ["DeepseekV4ForCausalLM"]), + ("ctx_use_kv_cache_manager_v2", True), + ("gen_kv_dtype", "bf16"), + ], +) +def test_fingerprint_covers_preference_inputs(field: str, value: object) -> None: + """Different preference inputs cannot share a static plan fingerprint.""" + plan = pcfg.resolve_plan(_disagg_yaml()) + before = plan["fingerprint"] + plan[field] = value + assert pcfg.plan_fingerprint(plan) != before diff --git a/tests/unittest/others/test_cache_transceiver_precheck_run.py b/tests/unittest/others/test_cache_transceiver_precheck_run.py index d392b0d5b99a..b54b51936f90 100644 --- a/tests/unittest/others/test_cache_transceiver_precheck_run.py +++ b/tests/unittest/others/test_cache_transceiver_precheck_run.py @@ -36,6 +36,7 @@ import os import sys import types +from pathlib import Path import pytest @@ -1230,6 +1231,56 @@ def test_deepseek_v4_auto_selects_kv_cache_manager_v2(self, api, tmp_path): assert rp.resolve_model_prefs(str(model_dir), side, cache_cfg) is True + @pytest.mark.parametrize("role", ["ctx", "gen"]) + @pytest.mark.parametrize( + "topology", + [ + "con1156_ctx2_dep8_gen1_dep8_eplb0_dspark3", + "con1456_ctx3_dep8_gen1_dep16_eplb0_dspark5", + ], + ) + def test_deepseek_v4_declared_architecture_without_checkpoint( + self, api: object, tmp_path: Path, role: str, topology: str + ) -> None: + """Both failing configurations resolve V2/PYTHON without model weights.""" + import yaml + + config_path = ( + Path(rp.__file__).parent.parent + / "disaggregated" + / (f"gb300_deepseek-v4-pro-dspark_agentx_{topology}_ccb-NIXL.yaml") + ) + cfg = yaml.safe_load(config_path.read_text()) + assert rp.pcfg.resolve_model_dir(cfg, llm_models_root=str(tmp_path)) is None + side = rp.pcfg.side_plan(rp.pcfg.resolve_plan(cfg), role) + cache_cfg = api.CacheTransceiverConfig(**side["cache_transceiver_config"]) + assert rp.resolve_model_prefs(None, side, cache_cfg) is True + assert cache_cfg.transceiver_runtime == "PYTHON" + + def test_resolved_contract_includes_pipelined_runtime_promotion(self, api: object) -> None: + """Record the final Python runtime when pipelining promotes an unset runtime.""" + cfg = { + "hardware": {"gpus_per_node": 1, "num_ctx_servers": 1, "num_gen_servers": 1}, + "worker_config": { + role: { + "cache_transceiver_config": { + "backend": "NIXL", + "transceiver_runtime": None, + "enable_pipelined_transfer": True, + }, + "kv_cache_config": {"use_kv_cache_manager_v2": False}, + } + for role in ("ctx", "gen") + }, + } + runner = rp.PrecheckRunner.__new__(rp.PrecheckRunner) + runner.plan = rp.pcfg.resolve_plan(cfg) + runner.role, runner.is_leader = "ctx", False + resolved = runner._resolve_transfer_config(dict(rp.pcfg.FALLBACK_KV_SHAPE)) + assert resolved.transceiver_runtime == "PYTHON" + assert runner.transfer_contract["ctx"]["runtime"] == "PYTHON" + assert runner.transfer_contract["gen"]["runtime"] == "PYTHON" + def test_enum_members(self, api): for enum, members in ( (api.DataType, ("FP8", "HALF", "BF16")), @@ -1300,3 +1351,131 @@ def test_make_request_shapes(self, api): assert gen_req.py_request_id == 13 py_gen = rp.make_request(False, rid=14, req_len=8, runtime="PYTHON", ctx_params=ctx_params) assert py_gen.py_disaggregated_params.request_type == "generation_only" + + +@pytest.mark.parametrize( + "model_dir,declared,registered,expected", + [ + (None, ["Declared"], True, True), + ("/checkpoint", ["Declared"], True, False), + (None, [], False, None), + (None, ["Unknown"], False, None), + ("/checkpoint", ["Declared"], False, None), + ], +) +def test_architecture_fallback_precedence( + monkeypatch: pytest.MonkeyPatch, + model_dir: str | None, + declared: list[str], + registered: bool, + expected: bool | None, +) -> None: + """Checkpoint identity wins; unknown or missing identity still fails closed.""" + checkpoint_cls = object() if model_dir and registered else None + declared_cls = object() if registered or model_dir else None + view = object() if model_dir else None + monkeypatch.setattr(rp, "_lookup_model_cls", lambda _: (checkpoint_cls, view)) + api = types.SimpleNamespace( + get_registered_model_class=lambda _: declared_cls, + TorchLlmArgs=lambda **kwargs: types.SimpleNamespace(**kwargs), + resolve_kv_cache_manager_v2_auto=lambda args, cls, config: cls is declared_cls, + ) + monkeypatch.setattr(rp, "load_internal_apis", lambda: api) + side = { + "architectures": declared, + "use_kv_cache_manager_v2": "auto", + "parallel": {"tp": 1, "pp": 1, "cp": 1}, + } + cache_cfg = types.SimpleNamespace(transceiver_runtime="PYTHON") + if expected is None: + with pytest.raises(RuntimeError, match="refusing to assume V1"): + rp.resolve_model_prefs(model_dir, side, cache_cfg) + else: + assert rp.resolve_model_prefs(model_dir, side, cache_cfg) is expected + + +@pytest.mark.parametrize("failure", ["resolution", "rank_contract", "api_import"]) +def test_setup_rejects_rank_divergence_before_construction( + monkeypatch: pytest.MonkeyPatch, failure: str +) -> None: + """All ranks synchronize local errors before entering GPU constructors.""" + runner = rp.PrecheckRunner.__new__(rp.PrecheckRunner) + runner.transfer_contract = {"ctx": {"use_v2": True}} + calls = [] + + def allgather(value: object) -> list: + """Emulate one remote rank disagreeing at the correct collective.""" + calls.append(value) + if len(calls) == 1: + return [value, "missing architecture" if failure == "resolution" else ""] + return [value, {"ctx": {"use_v2": False}}] + + runner.comm = types.SimpleNamespace(allgather=allgather) + if failure != "api_import": + runner._resolve_transfer_config = lambda shape: object() + + def load_api() -> object: + """Fail the real preparation import when testing rank-local errors.""" + if failure == "api_import": + raise ImportError("native binding unavailable") + return types.SimpleNamespace() + + monkeypatch.setattr(rp, "load_internal_apis", load_api) + with pytest.raises( + RuntimeError, + match=( + "configuration resolution" + if failure != "rank_contract" + else "contract mismatch within instance" + ), + ): + runner.setup({}, 32) + assert len(calls) == (2 if failure == "rank_contract" else 1) + + +@pytest.mark.parametrize("role", ["ctx", "gen"]) +def test_resolved_contract_preserves_role_asymmetry( + monkeypatch: pytest.MonkeyPatch, role: str +) -> None: + """Both peers derive the same full contract despite role and path differences.""" + import precheck_config as pcfg + + cfg = { + "metadata": {"architectures": ["Example"]}, + "hardware": {"gpus_per_node": 1, "num_ctx_servers": 1, "num_gen_servers": 1}, + "worker_config": { + side: { + "cache_transceiver_config": {"backend": "DEFAULT", "transceiver_runtime": "PYTHON"}, + "kv_cache_config": { + "use_kv_cache_manager_v2": side == "gen", + "dtype": "half" if side == "ctx" else "fp16", + }, + } + for side in ("ctx", "gen") + }, + } + + def make_cache_config(**kwargs: object) -> object: + """Stand in for external pydantic config with its backend resolver surface.""" + return types.SimpleNamespace(**kwargs, _resolve_default_backend=lambda: ("NIXL", None)) + + monkeypatch.setattr( + rp, + "load_internal_apis", + lambda: types.SimpleNamespace( + CacheTransceiverConfig=make_cache_config, + resolve_cache_transceiver_config=lambda cfg: None, + get_registered_model_class=lambda name: object(), + ), + ) + runner = rp.PrecheckRunner.__new__(rp.PrecheckRunner) + runner.plan = pcfg.resolve_plan(cfg) + runner.role, runner.is_leader = role, False + shape = dict(pcfg.FALLBACK_KV_SHAPE, source="different local path") + resolved = runner._resolve_transfer_config(shape) + assert runner.use_v2 is (role == "gen") + assert resolved.transceiver_runtime == "PYTHON" + expected = {"runtime": "PYTHON", "backend": "NIXL", "kv_dtype": "fp16"} + assert runner.transfer_contract["ctx"] == dict(expected, use_v2=False) + assert runner.transfer_contract["gen"] == dict(expected, use_v2=True) + assert "source" not in runner.transfer_contract["kv_shape"] From 374e48c65ecebc869926f79080a296098dddfb92 Mon Sep 17 00:00:00 2001 From: Chien-Chun Hung <2679986+chienchunhung@users.noreply.github.com> Date: Fri, 2 Oct 2026 16:19:11 -0700 Subject: [PATCH 2/4] fix: clarify precheck architecture and checkpoint status Signed-off-by: Chien-Chun Hung <2679986+chienchunhung@users.noreply.github.com> --- .../cache_transceiver_precheck/README.md | 19 ++- .../precheck_config.py | 29 ++-- .../run_precheck.py | 30 +++- ...dep8_gen1_dep8_eplb0_dspark3_ccb-NIXL.yaml | 2 +- ...ep8_gen1_dep16_eplb0_dspark5_ccb-NIXL.yaml | 2 +- ...dep8_gen1_dep8_eplb0_dspark3_ccb-NIXL.yaml | 2 +- ...ep8_gen1_dep16_eplb0_dspark5_ccb-NIXL.yaml | 2 +- .../test_cache_transceiver_precheck_e2e.py | 12 +- .../test_cache_transceiver_precheck_config.py | 54 +++++--- .../test_cache_transceiver_precheck_run.py | 129 ++++++++++++++++-- 10 files changed, 220 insertions(+), 61 deletions(-) diff --git a/tests/scripts/perf-sanity/cache_transceiver_precheck/README.md b/tests/scripts/perf-sanity/cache_transceiver_precheck/README.md index 0e4aa14d991b..f6ece16b1f2f 100644 --- a/tests/scripts/perf-sanity/cache_transceiver_precheck/README.md +++ b/tests/scripts/perf-sanity/cache_transceiver_precheck/README.md @@ -16,7 +16,7 @@ starts. | Same UCX env vars (incl. the `unset UCX_TLS` cases) | `jenkins/scripts/perf/submit.py` builds the precheck commands from the **same** `ucx_tls_cmd` + `$CTX/GEN_WORKER_ENV_VARS` strings as the worker steps; `slurm_precheck_run.sh` sources the same `slurm_env_setup.sh` (the `UCX_TLS=tcp` fixup) as `slurm_run.sh`. | | Same instance count / parallelism | One precheck `srun` per ctx/gen server with the same `-N/--ntasks/--ntasks-per-node/--mpi=pmix` and the same node slices (`-w`) as the real server steps (`slurm_launch_draft.sh`). TP/PP/CP/attention-DP come from the same `worker_config`. | | Same transceiver config | `CacheTransceiverConfig(**yaml["worker_config"][role]["cache_transceiver_config"])` — the yaml block is passed through verbatim (backend, `max_tokens_in_buffer`, timeouts, ...). | -| Same KV cache manager version + transceiver runtime | The launch generator forwards the real test's `LLM_MODELS_ROOT` into every precheck process. Explicit per-side `kv_cache_config.use_kv_cache_manager_v2` wins; absent means "auto" and requires a registered model class, then resolves via `get_preferred_kv_cache_manager_version()`. `transceiver_runtime: auto` resolves via `get_preferred_transceiver_runtime()` (NIXL-gated) — both through the same llm_utils code serving uses. The model class comes from the checkpoint; when no checkpoint directory is resolved, `metadata.architectures` supplies a registered architecture hint. An unresolved manager-version `auto` setting fails with INIT_ERROR instead of silently assuming V1. V2 requires the Python transceiver (the C++ one only supports V1); a V2+CPP combination also fails fast. | +| Same KV cache manager version + transceiver runtime | The launch generator forwards the real test's `LLM_MODELS_ROOT` into every precheck process. Explicit per-side `kv_cache_config.use_kv_cache_manager_v2` wins; absent means "auto" and requires a registered model class, then resolves via `get_preferred_kv_cache_manager_version()`. `transceiver_runtime: auto` resolves via `get_preferred_transceiver_runtime()` (NIXL-gated) — both through the same llm_utils code serving uses. The model class comes from the checkpoint; when no checkpoint directory is resolved, `metadata.architecture` supplies a registered architecture hint. An unresolved manager-version `auto` setting fails with INIT_ERROR instead of silently assuming V1. V2 requires the Python transceiver (the C++ one only supports V1); a V2+CPP combination also fails fast. | Asymmetric layouts (ctx dep4 → gen dep16, ctx pp8 → gen tp32, ...) are supported: data is seeded per (request, **global** layer) and constant along @@ -28,12 +28,13 @@ For checkpoints unavailable to the network precheck, declare the model architect ```yaml metadata: - architectures: [DeepseekV4ForCausalLM] + architecture: DeepseekV4ForCausalLM ``` -The first declaration is looked up in the model registry. A resolved checkpoint -always takes precedence; metadata does not replace an unknown or invalid -checkpoint architecture. Model hooks still run through the serving resolvers, +The hint accepts one nonempty architecture name, which is looked up in the +model registry; lists are rejected. A resolved checkpoint always takes +precedence; metadata does not replace an unknown or invalid checkpoint +architecture. Model hooks still run through the serving resolvers, and a hook that needs unavailable checkpoint fields fails closed. Without a checkpoint, the precheck transfers a generic synthetic KV pool. @@ -114,6 +115,14 @@ csv/ctx_/_.csv # Python transceiver per-task perf # (KVSendTask throughput on the ctx side) ``` +Every status JSON written by the runner includes `model_dir_resolved` and +`synthetic_kv_shape`, including intermediate records and failure results. +`model_dir_resolved` means a directory containing `config.json` was resolved; +it does not establish that model weights are available or valid. +`synthetic_kv_shape` is true for a missing checkpoint, a simplified layout, +or a generic fallback shape, including fallback from a resolved directory. +These fields qualify the transport result without changing the gate's verdict. + ## Manual runs ```bash diff --git a/tests/scripts/perf-sanity/cache_transceiver_precheck/precheck_config.py b/tests/scripts/perf-sanity/cache_transceiver_precheck/precheck_config.py index ccc45396eff4..ce6c6a2af30f 100644 --- a/tests/scripts/perf-sanity/cache_transceiver_precheck/precheck_config.py +++ b/tests/scripts/perf-sanity/cache_transceiver_precheck/precheck_config.py @@ -339,26 +339,25 @@ def gate_library_content(draft_launch_sh, llm_src): ) -def declared_architectures(cfg: Mapping[str, Any]) -> list[str]: - """Read architecture hints for prechecks without a local checkpoint. +def declared_architecture(cfg: Mapping[str, Any]) -> str | None: + """Read one architecture hint for prechecks without a local checkpoint. Args: cfg: Parsed disaggregated benchmark YAML. Returns: - Architecture names in declared priority order, or an empty list. + The stripped architecture name, or None when the hint is omitted. Raises: - ValueError: If the declaration is not a string or list of nonempty strings. + ValueError: If a supplied declaration is not a nonempty string. """ - value = (cfg.get("metadata") or {}).get("architectures", []) - if isinstance(value, str): - value = [value] - if not isinstance(value, list) or any( - not isinstance(name, str) or not name.strip() for name in value - ): - raise ValueError("metadata.architectures must be a string or list of nonempty strings") - return [name.strip() for name in value] + metadata = cfg.get("metadata") or {} + if "architecture" not in metadata: + return None + value = metadata["architecture"] + if not isinstance(value, str) or not value.strip(): + raise ValueError("metadata.architecture must be a nonempty string") + return value.strip() def normalized_kv_dtype(dtype: str) -> str: @@ -473,7 +472,7 @@ def resolve_plan(cfg, benchmark_mode="e2e"): ), "rendezvous_timeout_s": rendezvous_timeout_s, "verify_data": bool(knobs["verify_data"]), - "architectures": declared_architectures(cfg), + "architecture": declared_architecture(cfg), } for role, side, xcvr in (("ctx", ctx_side, ctx_xcvr), ("gen", gen_side, gen_xcvr)): kv_cfg = side.get("kv_cache_config") or {} @@ -492,7 +491,7 @@ def resolve_plan(cfg, benchmark_mode="e2e"): def plan_fingerprint(plan): """Stable string both sides must agree on before transferring.""" keys = ( - "architectures", + "architecture", "ctx_use_kv_cache_manager_v2", "gen_use_kv_cache_manager_v2", "ctx_kv_dtype", @@ -519,7 +518,7 @@ def side_plan(plan, role): """Per-role view: this role's parallelism + transceiver/kv config.""" return { "role": role, - "architectures": plan["architectures"], + "architecture": plan["architecture"], "parallel": plan[role], "cache_transceiver_config": plan[f"{role}_cache_transceiver_config"], "kv_dtype": plan[f"{role}_kv_dtype"], diff --git a/tests/scripts/perf-sanity/cache_transceiver_precheck/run_precheck.py b/tests/scripts/perf-sanity/cache_transceiver_precheck/run_precheck.py index b72756319c35..ec6a25a1d511 100644 --- a/tests/scripts/perf-sanity/cache_transceiver_precheck/run_precheck.py +++ b/tests/scripts/perf-sanity/cache_transceiver_precheck/run_precheck.py @@ -392,15 +392,15 @@ class and adopt its get_preferred_kv_cache_manager_version() value api = load_internal_apis() model_cls, hf_view = _lookup_model_cls(model_dir) - declared = side.get("architectures", []) + declared = side.get("architecture") if model_dir is None and declared: - model_cls = api.get_registered_model_class(declared[0]) + model_cls = api.get_registered_model_class(declared) setting = side["use_kv_cache_manager_v2"] if setting == "auto" and model_cls is None: raise RuntimeError( "use_kv_cache_manager_v2 is 'auto', but the precheck could not resolve " f"a registered model class from model_dir={model_dir!r} or " - f"metadata.architectures={declared!r}; refusing to assume V1. " + f"metadata.architecture={declared!r}; refusing to assume V1. " "Check LLM_MODELS_ROOT/checkpoint staging or declare the model architecture." ) @@ -455,7 +455,6 @@ def build_kv_cache_manager(kv_shape, plan, side, mapping, max_req_len, use_v2): dtype = dtype_map[pcfg.normalized_kv_dtype(dtype_str)] if dtype_str not in dtype_map: print(f"[precheck] kv dtype {dtype_str!r} not mapped, using BF16", flush=True) - dtype = api.DataType.BF16 spec_config = None if side["num_nextn_predict_layers"] > 0: @@ -767,11 +766,12 @@ class StatusRecorder: cases + the in-flight failure on disk. """ - def __init__(self, work_dir, role, server_idx, is_leader): + def __init__(self, work_dir, role, server_idx, is_leader, metadata=None): self.role = role self.server_idx = server_idx self.is_leader = is_leader self.cases = [] + self.metadata = dict(metadata or {}) status_dir = os.path.join(work_dir, "status") if is_leader: os.makedirs(status_dir, exist_ok=True) @@ -807,6 +807,7 @@ def _flush(self, final, extra=None): "overall": overall, "cases": self.cases, "env": self.env, + **self.metadata, } if extra: doc.update(extra) @@ -929,7 +930,22 @@ def __init__(self, args, plan, side, comm): self.rank = comm.Get_rank() self.is_leader = self.rank == 0 self.work_dir = args.work_dir - self.recorder = StatusRecorder(self.work_dir, self.role, self.server_idx, self.is_leader) + model_dir_resolved = plan.get("_model_dir") is not None + kv_shape = plan.get("_kv_shape", {}) + self.recorder = StatusRecorder( + self.work_dir, + self.role, + self.server_idx, + self.is_leader, + metadata={ + "model_dir_resolved": model_dir_resolved, + "synthetic_kv_shape": ( + not model_dir_resolved + or bool(kv_shape.get("simplified")) + or kv_shape.get("source") == "fallback" + ), + }, + ) self.zmq_ctx = None self.kvm = None self.xcvr = None @@ -999,7 +1015,7 @@ def _resolve_transfer_config(self, kv_shape: dict) -> object: print( f"[precheck {self.role}_{self.server_idx}] synthetic KV pool; " f"model_dir={self.plan.get('_model_dir')!r}, " - f"architectures={self.plan['architectures']!r}. " + f"architecture={self.plan['architecture']!r}. " "This checks cache transport; serving still requires a staged checkpoint.", flush=True, ) diff --git a/tests/scripts/perf-sanity/disaggregated/gb300_deepseek-v4-pro-dspark_agentx_con1156_ctx2_dep8_gen1_dep8_eplb0_dspark3_ccb-NIXL.yaml b/tests/scripts/perf-sanity/disaggregated/gb300_deepseek-v4-pro-dspark_agentx_con1156_ctx2_dep8_gen1_dep8_eplb0_dspark3_ccb-NIXL.yaml index 8ee1d9bf96d4..63dfd8b70166 100644 --- a/tests/scripts/perf-sanity/disaggregated/gb300_deepseek-v4-pro-dspark_agentx_con1156_ctx2_dep8_gen1_dep8_eplb0_dspark3_ccb-NIXL.yaml +++ b/tests/scripts/perf-sanity/disaggregated/gb300_deepseek-v4-pro-dspark_agentx_con1156_ctx2_dep8_gen1_dep8_eplb0_dspark3_ccb-NIXL.yaml @@ -1,5 +1,5 @@ metadata: - architectures: [DeepseekV4ForCausalLM] + architecture: DeepseekV4ForCausalLM model_name: deepseek_v4_pro_nvfp4_dspark precision: fp4 model_dir_name: DeepSeek-V4-Pro-nvfp4-DSpark diff --git a/tests/scripts/perf-sanity/disaggregated/gb300_deepseek-v4-pro-dspark_agentx_con1456_ctx3_dep8_gen1_dep16_eplb0_dspark5_ccb-NIXL.yaml b/tests/scripts/perf-sanity/disaggregated/gb300_deepseek-v4-pro-dspark_agentx_con1456_ctx3_dep8_gen1_dep16_eplb0_dspark5_ccb-NIXL.yaml index b8c756cb4937..ddef64d8567e 100644 --- a/tests/scripts/perf-sanity/disaggregated/gb300_deepseek-v4-pro-dspark_agentx_con1456_ctx3_dep8_gen1_dep16_eplb0_dspark5_ccb-NIXL.yaml +++ b/tests/scripts/perf-sanity/disaggregated/gb300_deepseek-v4-pro-dspark_agentx_con1456_ctx3_dep8_gen1_dep16_eplb0_dspark5_ccb-NIXL.yaml @@ -1,5 +1,5 @@ metadata: - architectures: [DeepseekV4ForCausalLM] + architecture: DeepseekV4ForCausalLM model_name: deepseek_v4_pro_nvfp4_dspark precision: fp4 model_dir_name: DeepSeek-V4-Pro-nvfp4-DSpark diff --git a/tests/scripts/perf/disaggregated/gb300_deepseek-v4-pro-dspark_agentx_con1156_ctx2_dep8_gen1_dep8_eplb0_dspark3_ccb-NIXL.yaml b/tests/scripts/perf/disaggregated/gb300_deepseek-v4-pro-dspark_agentx_con1156_ctx2_dep8_gen1_dep8_eplb0_dspark3_ccb-NIXL.yaml index 8ee1d9bf96d4..63dfd8b70166 100644 --- a/tests/scripts/perf/disaggregated/gb300_deepseek-v4-pro-dspark_agentx_con1156_ctx2_dep8_gen1_dep8_eplb0_dspark3_ccb-NIXL.yaml +++ b/tests/scripts/perf/disaggregated/gb300_deepseek-v4-pro-dspark_agentx_con1156_ctx2_dep8_gen1_dep8_eplb0_dspark3_ccb-NIXL.yaml @@ -1,5 +1,5 @@ metadata: - architectures: [DeepseekV4ForCausalLM] + architecture: DeepseekV4ForCausalLM model_name: deepseek_v4_pro_nvfp4_dspark precision: fp4 model_dir_name: DeepSeek-V4-Pro-nvfp4-DSpark diff --git a/tests/scripts/perf/disaggregated/gb300_deepseek-v4-pro-dspark_agentx_con1456_ctx3_dep8_gen1_dep16_eplb0_dspark5_ccb-NIXL.yaml b/tests/scripts/perf/disaggregated/gb300_deepseek-v4-pro-dspark_agentx_con1456_ctx3_dep8_gen1_dep16_eplb0_dspark5_ccb-NIXL.yaml index b8c756cb4937..ddef64d8567e 100644 --- a/tests/scripts/perf/disaggregated/gb300_deepseek-v4-pro-dspark_agentx_con1456_ctx3_dep8_gen1_dep16_eplb0_dspark5_ccb-NIXL.yaml +++ b/tests/scripts/perf/disaggregated/gb300_deepseek-v4-pro-dspark_agentx_con1456_ctx3_dep8_gen1_dep16_eplb0_dspark5_ccb-NIXL.yaml @@ -1,5 +1,5 @@ metadata: - architectures: [DeepseekV4ForCausalLM] + architecture: DeepseekV4ForCausalLM model_name: deepseek_v4_pro_nvfp4_dspark precision: fp4 model_dir_name: DeepSeek-V4-Pro-nvfp4-DSpark diff --git a/tests/unittest/disaggregated/test_cache_transceiver_precheck_e2e.py b/tests/unittest/disaggregated/test_cache_transceiver_precheck_e2e.py index cc6aa0b2d796..f6bc64faa020 100644 --- a/tests/unittest/disaggregated/test_cache_transceiver_precheck_e2e.py +++ b/tests/unittest/disaggregated/test_cache_transceiver_precheck_e2e.py @@ -256,7 +256,7 @@ def _read_status(work_dir, name): return text, doc -def _assert_all_passed(work_dir, launched): +def _assert_all_passed(work_dir, launched, *, model_dir_resolved=True, synthetic_kv_shape=False): failures = [] for name, proc, log_path in launched: if proc.returncode != 0: @@ -271,6 +271,8 @@ def _assert_all_passed(work_dir, launched): assert doc["overall"] == "PASS" assert doc["transceiver_runtime"] == "PYTHON" assert doc["kv_cache_manager"] == "V2" + assert doc["model_dir_resolved"] is model_dir_resolved + assert doc["synthetic_kv_shape"] is synthetic_kv_shape # The Python transceiver records bandwidth on the ctx (sender) leader # via PerfLogManager CSVs in TRTLLM_KVCACHE_TIME_OUTPUT_PATH. Asserting # it here ties the whole chain (perf_logger naming -> CSV -> parser) @@ -352,6 +354,8 @@ def test_precheck_fails_fast_on_fingerprint_mismatch(tmp_path): text, doc = _read_status(work_dir, name) assert text.startswith(f"FAIL {name}"), f"{name} status: {text}" assert doc["overall"] == "FAIL" + assert doc["model_dir_resolved"] is True + assert doc["synthetic_kv_shape"] is False # The gen driver saw the ctx abort its handshake: the reason must name it. text, _ = _read_status(work_dir, "gen_0") assert "fingerprint" in text or "abort" in text.lower() @@ -362,14 +366,14 @@ def test_precheck_declared_architecture_without_checkpoint(tmp_path: Path) -> No """Missing weights still allow actual V2/Python transfers after auto resolution.""" pytest.importorskip("mpi4py") cfg = _disagg_yaml(1, 1, 1, 1) - cfg["metadata"] = {"model_dir_name": "unstaged", "architectures": ["DeepseekV4ForCausalLM"]} + cfg["metadata"] = {"model_dir_name": "unstaged", "architecture": "DeepseekV4ForCausalLM"} for side in cfg["worker_config"].values(): side["kv_cache_config"].pop("use_kv_cache_manager_v2") side["cache_transceiver_config"]["transceiver_runtime"] = "auto" config_path, models_root = _write_inputs(tmp_path, cfg) work_dir, launched = _launch_instances(tmp_path, _jobs(cfg, config_path), models_root) _wait_all(launched) - _assert_all_passed(work_dir, launched) + _assert_all_passed(work_dir, launched, model_dir_resolved=False, synthetic_kv_shape=True) @pytest.mark.timeout(300) @@ -389,4 +393,6 @@ def test_precheck_rejects_inconsistent_checkpoint_visibility(tmp_path: Path) -> text, doc = _read_status(work_dir, name) assert "resolved transfer contract mismatch" in text assert doc["overall"] == "FAIL" + assert doc["model_dir_resolved"] is (name == "ctx_0") + assert doc["synthetic_kv_shape"] is (name == "gen_0") assert not any(case["status"] == "PASS" for case in doc["cases"]) diff --git a/tests/unittest/others/test_cache_transceiver_precheck_config.py b/tests/unittest/others/test_cache_transceiver_precheck_config.py index 80e2b7c17764..698127cf66d7 100644 --- a/tests/unittest/others/test_cache_transceiver_precheck_config.py +++ b/tests/unittest/others/test_cache_transceiver_precheck_config.py @@ -1445,33 +1445,55 @@ def test_python_transceiver_bandwidth_csv(tmp_path): @pytest.mark.parametrize( - "value,expected", + "metadata,expected", [ - ([], []), - (" DeepseekV4ForCausalLM ", ["DeepseekV4ForCausalLM"]), - (["DeepseekV4ForCausalLM"], ["DeepseekV4ForCausalLM"]), + ({}, None), + ({"architecture": "DeepseekV4ForCausalLM"}, "DeepseekV4ForCausalLM"), + ({"architecture": " DeepseekV4ForCausalLM "}, "DeepseekV4ForCausalLM"), ], ) -def test_declared_architectures(value: object, expected: list[str]) -> None: - """Normalize metadata and retain it in the shared and per-role plans.""" - cfg = _disagg_yaml(metadata={"architectures": value}) +def test_declared_architecture(metadata: dict, expected: str | None) -> None: + """Retain one optional, normalized architecture in both roles and the fingerprint.""" + cfg = _disagg_yaml(metadata=metadata) + assert pcfg.declared_architecture(cfg) == expected plan = pcfg.resolve_plan(cfg) - assert plan["architectures"] == expected - assert pcfg.side_plan(plan, "ctx")["architectures"] == expected - assert json.loads(plan["fingerprint"])["architectures"] == expected + assert plan["architecture"] == expected + for role in ("ctx", "gen"): + assert pcfg.side_plan(plan, role)["architecture"] == expected + assert json.loads(plan["fingerprint"])["architecture"] == expected + if expected is not None: + normalized = pcfg.resolve_plan(_disagg_yaml(metadata={"architecture": expected})) + assert normalized["fingerprint"] == plan["fingerprint"] -@pytest.mark.parametrize("value", [None, 42, {}, [""], [" "], [None], [42]]) -def test_invalid_declared_architectures(value: object) -> None: - """Malformed hints fail early rather than selecting an unintended class.""" - with pytest.raises(ValueError, match="metadata.architectures"): - pcfg.resolve_plan(_disagg_yaml(metadata={"architectures": value})) +@pytest.mark.parametrize( + "value", + [ + None, + "", + " \t\n", + 42, + {}, + True, + [], + [""], + [" "], + [None], + [42], + ["DeepseekV4ForCausalLM"], + ["Unknown", "DeepseekV4ForCausalLM"], + ], +) +def test_invalid_declared_architecture(value: object) -> None: + """Reject malformed scalars and every list, including a valid later entry.""" + with pytest.raises(ValueError, match=r"metadata\.architecture"): + pcfg.resolve_plan(_disagg_yaml(metadata={"architecture": value})) @pytest.mark.parametrize( "field,value", [ - ("architectures", ["DeepseekV4ForCausalLM"]), + ("architecture", "DeepseekV4ForCausalLM"), ("ctx_use_kv_cache_manager_v2", True), ("gen_kv_dtype", "bf16"), ], diff --git a/tests/unittest/others/test_cache_transceiver_precheck_run.py b/tests/unittest/others/test_cache_transceiver_precheck_run.py index b54b51936f90..27182b8ec11f 100644 --- a/tests/unittest/others/test_cache_transceiver_precheck_run.py +++ b/tests/unittest/others/test_cache_transceiver_precheck_run.py @@ -217,16 +217,104 @@ def _read_status(tmp_path, name): def test_recorder_pass(tmp_path): - rec = rp.StatusRecorder(str(tmp_path), "gen", 0, is_leader=True) + metadata = {"model_dir_resolved": True, "synthetic_kv_shape": False} + rec = rp.StatusRecorder(str(tmp_path), "gen", 0, is_leader=True, metadata=metadata) + metadata["model_dir_resolved"] = False rec.record("ctx_0", 1024, "PASS") text, doc = _read_status(tmp_path, "gen_0") assert text.startswith("RUNNING") # not final yet: a SIGKILL must not read as PASS assert doc["overall"] == "RUNNING" + assert doc["model_dir_resolved"] is True + assert doc["synthetic_kv_shape"] is False rec.finalize(extra={"transceiver_runtime": "CPP"}) text, doc = _read_status(tmp_path, "gen_0") assert text.startswith("PASS gen_0") assert doc["overall"] == "PASS" assert doc["transceiver_runtime"] == "CPP" + assert doc["model_dir_resolved"] is True + assert doc["synthetic_kv_shape"] is False + + +@pytest.mark.parametrize( + "checkpoint,case_status,model_dir_resolved,synthetic_kv_shape", + [ + ("unresolved", "INIT_ERROR", False, True), + ("gqa", "PASS", True, False), + ("sparse", "TIMEOUT", True, True), + ("malformed", "TRANSFER_ERROR", True, True), + ], +) +def test_runner_persists_checkpoint_metadata( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, + checkpoint: str, + case_status: str, + model_dir_resolved: bool, + synthetic_kv_shape: bool, +) -> None: + """Carry actual checkpoint resolution through successful and failed status writes.""" + models_root = tmp_path / "models" + model_dir = models_root / "model" + if checkpoint != "unresolved": + model_dir.mkdir(parents=True) + hf_config = { + "num_hidden_layers": 2, + "num_attention_heads": 4, + "num_key_value_heads": 2, + "head_dim": 64, + } + if checkpoint == "sparse": + hf_config["index_head_dim"] = 128 + config_text = "{" if checkpoint == "malformed" else json.dumps(hf_config) + (model_dir / "config.json").write_text(config_text) + monkeypatch.setenv("LLM_MODELS_ROOT", str(models_root)) + config_path = tmp_path / "precheck.yaml" + config_path.write_text( + json.dumps( + { + "metadata": {"model_dir_name": "model"}, + "hardware": {"gpus_per_node": 1, "num_ctx_servers": 1, "num_gen_servers": 1}, + "worker_config": { + role: { + "tensor_parallel_size": 1, + "cache_transceiver_config": {"backend": "NIXL"}, + } + for role in ("ctx", "gen") + }, + } + ) + ) + args = rp.parse_args( + [ + "--role", + "gen", + "--server-idx", + "0", + "--config", + str(config_path), + "--work-dir", + str(tmp_path), + ] + ) + plan, side = rp.load_plan(args) + monkeypatch.setitem(sys.modules, "mpi4py", types.SimpleNamespace(MPI=None)) + runner = rp.PrecheckRunner(args, plan, side, types.SimpleNamespace(Get_rank=lambda: 0)) + runner.recorder.record("ctx_0", 64, case_status, "example outcome") + _, doc = _read_status(tmp_path, "gen_0") + assert doc["overall"] == ("RUNNING" if case_status == "PASS" else "FAIL") + assert doc["model_dir_resolved"] is model_dir_resolved + assert doc["synthetic_kv_shape"] is synthetic_kv_shape + runner.recorder.finalize( + extra={"transceiver_runtime": "PYTHON", "kv_cache_manager": "V2", "per_gpu_bw_gbps": 2.5} + ) + _, doc = _read_status(tmp_path, "gen_0") + assert doc["overall"] == ("PASS" if case_status == "PASS" else "FAIL") + assert doc["cases"][0]["status"] == case_status + assert doc["model_dir_resolved"] is model_dir_resolved + assert doc["synthetic_kv_shape"] is synthetic_kv_shape + assert doc["transceiver_runtime"] == "PYTHON" + assert doc["kv_cache_manager"] == "V2" + assert doc["per_gpu_bw_gbps"] == 2.5 def test_recorder_failure_summary_first_line_only(tmp_path): @@ -1224,6 +1312,7 @@ def test_deepseek_v4_auto_selects_kv_cache_manager_v2(self, api, tmp_path): json.dumps({"architectures": ["DeepseekV4ForCausalLM"]}) ) side = { + "architecture": "LlamaForCausalLM", "use_kv_cache_manager_v2": "auto", "parallel": {"tp": 1, "pp": 1, "cp": 1}, } @@ -1253,6 +1342,7 @@ def test_deepseek_v4_declared_architecture_without_checkpoint( cfg = yaml.safe_load(config_path.read_text()) assert rp.pcfg.resolve_model_dir(cfg, llm_models_root=str(tmp_path)) is None side = rp.pcfg.side_plan(rp.pcfg.resolve_plan(cfg), role) + assert side["architecture"] == "DeepseekV4ForCausalLM" cache_cfg = api.CacheTransceiverConfig(**side["cache_transceiver_config"]) assert rp.resolve_model_prefs(None, side, cache_cfg) is True assert cache_cfg.transceiver_runtime == "PYTHON" @@ -1356,17 +1446,17 @@ def test_make_request_shapes(self, api): @pytest.mark.parametrize( "model_dir,declared,registered,expected", [ - (None, ["Declared"], True, True), - ("/checkpoint", ["Declared"], True, False), - (None, [], False, None), - (None, ["Unknown"], False, None), - ("/checkpoint", ["Declared"], False, None), + (None, "Declared", True, True), + ("/checkpoint", "Declared", True, False), + (None, None, False, None), + (None, "Unknown", False, None), + ("/checkpoint", "Declared", False, None), ], ) def test_architecture_fallback_precedence( monkeypatch: pytest.MonkeyPatch, model_dir: str | None, - declared: list[str], + declared: str | None, registered: bool, expected: bool | None, ) -> None: @@ -1374,15 +1464,29 @@ def test_architecture_fallback_precedence( checkpoint_cls = object() if model_dir and registered else None declared_cls = object() if registered or model_dir else None view = object() if model_dir else None + registry_calls = [] + resolver_calls = [] + + def get_registered_model_class(name: str) -> object | None: + """Record the full scalar used to select the exact declared class.""" + registry_calls.append(name) + assert name == declared + return declared_cls + + def resolve_v2(args: object, cls: object, config: object) -> bool: + """Capture the identity and checkpoint view passed to serving's resolver.""" + resolver_calls.append((cls, config)) + return cls is declared_cls + monkeypatch.setattr(rp, "_lookup_model_cls", lambda _: (checkpoint_cls, view)) api = types.SimpleNamespace( - get_registered_model_class=lambda _: declared_cls, + get_registered_model_class=get_registered_model_class, TorchLlmArgs=lambda **kwargs: types.SimpleNamespace(**kwargs), - resolve_kv_cache_manager_v2_auto=lambda args, cls, config: cls is declared_cls, + resolve_kv_cache_manager_v2_auto=resolve_v2, ) monkeypatch.setattr(rp, "load_internal_apis", lambda: api) side = { - "architectures": declared, + "architecture": declared, "use_kv_cache_manager_v2": "auto", "parallel": {"tp": 1, "pp": 1, "cp": 1}, } @@ -1390,8 +1494,11 @@ def test_architecture_fallback_precedence( if expected is None: with pytest.raises(RuntimeError, match="refusing to assume V1"): rp.resolve_model_prefs(model_dir, side, cache_cfg) + assert not resolver_calls else: assert rp.resolve_model_prefs(model_dir, side, cache_cfg) is expected + assert resolver_calls == [(checkpoint_cls if model_dir else declared_cls, view)] + assert registry_calls == ([declared] if model_dir is None and declared else []) @pytest.mark.parametrize("failure", ["resolution", "rank_contract", "api_import"]) @@ -1441,7 +1548,7 @@ def test_resolved_contract_preserves_role_asymmetry( import precheck_config as pcfg cfg = { - "metadata": {"architectures": ["Example"]}, + "metadata": {"architecture": "Example"}, "hardware": {"gpus_per_node": 1, "num_ctx_servers": 1, "num_gen_servers": 1}, "worker_config": { side: { From 031447e6cf37f64c2a27e2291cee8dcbf1c6dc18 Mon Sep 17 00:00:00 2001 From: Chien-Chun Hung <2679986+chienchunhung@users.noreply.github.com> Date: Fri, 2 Oct 2026 18:26:32 -0700 Subject: [PATCH 3/4] fix: declare architecture for all DeepSeek V4 prechecks Signed-off-by: Chien-Chun Hung <2679986+chienchunhung@users.noreply.github.com> --- ..._dep4_gen1_dep32_eplb384_mtp3_ccb-NIXL.yaml | 1 + ...2_dep4_gen1_dep8_eplb384_mtp1_ccb-NIXL.yaml | 1 + ..._dep4_gen1_dep16_eplb384_mtp3_ccb-NIXL.yaml | 1 + ...tx1_dep4_gen4_tep8_eplb0_mtp3_ccb-NIXL.yaml | 1 + ...7_dep4_gen1_dep8_eplb384_mtp3_ccb-NIXL.yaml | 1 + ...tx1_dep4_gen5_tep4_eplb0_mtp3_ccb-NIXL.yaml | 1 + ..._dep4_gen1_dep32_eplb384_mtp3_ccb-NIXL.yaml | 1 + ...7_dep4_gen1_dep8_eplb384_mtp3_ccb-NIXL.yaml | 1 + ...tx1_dep4_gen5_tep4_eplb0_mtp3_ccb-NIXL.yaml | 1 + .../test_cache_transceiver_precheck_run.py | 18 ++++++++---------- 10 files changed, 17 insertions(+), 10 deletions(-) diff --git a/tests/scripts/perf-sanity/disaggregated/gb300_deepseek-v4-pro-fp4_8k1k_con180_ctx3_dep4_gen1_dep32_eplb384_mtp3_ccb-NIXL.yaml b/tests/scripts/perf-sanity/disaggregated/gb300_deepseek-v4-pro-fp4_8k1k_con180_ctx3_dep4_gen1_dep32_eplb384_mtp3_ccb-NIXL.yaml index 738144721554..8d651ea090c9 100644 --- a/tests/scripts/perf-sanity/disaggregated/gb300_deepseek-v4-pro-fp4_8k1k_con180_ctx3_dep4_gen1_dep32_eplb384_mtp3_ccb-NIXL.yaml +++ b/tests/scripts/perf-sanity/disaggregated/gb300_deepseek-v4-pro-fp4_8k1k_con180_ctx3_dep4_gen1_dep32_eplb384_mtp3_ccb-NIXL.yaml @@ -1,4 +1,5 @@ metadata: + architecture: DeepseekV4ForCausalLM model_name: deepseek_v4_pro_fp4 precision: fp4 model_dir_name: DeepSeek-V4-Pro diff --git a/tests/scripts/perf-sanity/disaggregated/gb300_deepseek-v4-pro-fp4_8k1k_con4301_ctx12_dep4_gen1_dep8_eplb384_mtp1_ccb-NIXL.yaml b/tests/scripts/perf-sanity/disaggregated/gb300_deepseek-v4-pro-fp4_8k1k_con4301_ctx12_dep4_gen1_dep8_eplb384_mtp1_ccb-NIXL.yaml index c2403cd570f3..fc373cf30250 100644 --- a/tests/scripts/perf-sanity/disaggregated/gb300_deepseek-v4-pro-fp4_8k1k_con4301_ctx12_dep4_gen1_dep8_eplb384_mtp1_ccb-NIXL.yaml +++ b/tests/scripts/perf-sanity/disaggregated/gb300_deepseek-v4-pro-fp4_8k1k_con4301_ctx12_dep4_gen1_dep8_eplb384_mtp1_ccb-NIXL.yaml @@ -1,4 +1,5 @@ metadata: + architecture: DeepseekV4ForCausalLM model_name: deepseek_v4_pro_fp4 precision: fp4 model_dir_name: DeepSeek-V4-Pro diff --git a/tests/scripts/perf-sanity/disaggregated/gb300_deepseek-v4-pro-fp4_8k1k_con666_ctx6_dep4_gen1_dep16_eplb384_mtp3_ccb-NIXL.yaml b/tests/scripts/perf-sanity/disaggregated/gb300_deepseek-v4-pro-fp4_8k1k_con666_ctx6_dep4_gen1_dep16_eplb384_mtp3_ccb-NIXL.yaml index 7cf9a7640195..f5981fdb3df1 100644 --- a/tests/scripts/perf-sanity/disaggregated/gb300_deepseek-v4-pro-fp4_8k1k_con666_ctx6_dep4_gen1_dep16_eplb384_mtp3_ccb-NIXL.yaml +++ b/tests/scripts/perf-sanity/disaggregated/gb300_deepseek-v4-pro-fp4_8k1k_con666_ctx6_dep4_gen1_dep16_eplb384_mtp3_ccb-NIXL.yaml @@ -1,4 +1,5 @@ metadata: + architecture: DeepseekV4ForCausalLM model_name: deepseek_v4_pro_fp4 precision: fp4 model_dir_name: DeepSeek-V4-Pro diff --git a/tests/scripts/perf-sanity/disaggregated/gb300_deepseek-v4-pro-fp4_8k1k_con8_ctx1_dep4_gen4_tep8_eplb0_mtp3_ccb-NIXL.yaml b/tests/scripts/perf-sanity/disaggregated/gb300_deepseek-v4-pro-fp4_8k1k_con8_ctx1_dep4_gen4_tep8_eplb0_mtp3_ccb-NIXL.yaml index b54449034d31..e87e69176935 100644 --- a/tests/scripts/perf-sanity/disaggregated/gb300_deepseek-v4-pro-fp4_8k1k_con8_ctx1_dep4_gen4_tep8_eplb0_mtp3_ccb-NIXL.yaml +++ b/tests/scripts/perf-sanity/disaggregated/gb300_deepseek-v4-pro-fp4_8k1k_con8_ctx1_dep4_gen4_tep8_eplb0_mtp3_ccb-NIXL.yaml @@ -1,4 +1,5 @@ metadata: + architecture: DeepseekV4ForCausalLM model_name: deepseek_v4_pro_fp4 precision: fp4 model_dir_name: DeepSeek-V4-Pro diff --git a/tests/scripts/perf/disaggregated/gb300_deepseek-v4-pro-fp4_8k1k_con1229_ctx7_dep4_gen1_dep8_eplb384_mtp3_ccb-NIXL.yaml b/tests/scripts/perf/disaggregated/gb300_deepseek-v4-pro-fp4_8k1k_con1229_ctx7_dep4_gen1_dep8_eplb384_mtp3_ccb-NIXL.yaml index aa44793dc4e6..054d627253d6 100644 --- a/tests/scripts/perf/disaggregated/gb300_deepseek-v4-pro-fp4_8k1k_con1229_ctx7_dep4_gen1_dep8_eplb384_mtp3_ccb-NIXL.yaml +++ b/tests/scripts/perf/disaggregated/gb300_deepseek-v4-pro-fp4_8k1k_con1229_ctx7_dep4_gen1_dep8_eplb384_mtp3_ccb-NIXL.yaml @@ -1,4 +1,5 @@ metadata: + architecture: DeepseekV4ForCausalLM model_name: deepseek_v4_pro_fp4 precision: fp4 model_dir_name: DeepSeek-V4-Pro diff --git a/tests/scripts/perf/disaggregated/gb300_deepseek-v4-pro-fp4_8k1k_con2_ctx1_dep4_gen5_tep4_eplb0_mtp3_ccb-NIXL.yaml b/tests/scripts/perf/disaggregated/gb300_deepseek-v4-pro-fp4_8k1k_con2_ctx1_dep4_gen5_tep4_eplb0_mtp3_ccb-NIXL.yaml index 7cde7a5f53cb..7c39dd0f09b8 100644 --- a/tests/scripts/perf/disaggregated/gb300_deepseek-v4-pro-fp4_8k1k_con2_ctx1_dep4_gen5_tep4_eplb0_mtp3_ccb-NIXL.yaml +++ b/tests/scripts/perf/disaggregated/gb300_deepseek-v4-pro-fp4_8k1k_con2_ctx1_dep4_gen5_tep4_eplb0_mtp3_ccb-NIXL.yaml @@ -1,4 +1,5 @@ metadata: + architecture: DeepseekV4ForCausalLM model_name: deepseek_v4_pro_fp4 precision: fp4 model_dir_name: DeepSeek-V4-Pro diff --git a/tests/scripts/perf/disaggregated/gb300_deepseek-v4-pro-fp4_8k1k_con666_ctx8_dep4_gen1_dep32_eplb384_mtp3_ccb-NIXL.yaml b/tests/scripts/perf/disaggregated/gb300_deepseek-v4-pro-fp4_8k1k_con666_ctx8_dep4_gen1_dep32_eplb384_mtp3_ccb-NIXL.yaml index 8383675405a9..7aeb8369d8b3 100644 --- a/tests/scripts/perf/disaggregated/gb300_deepseek-v4-pro-fp4_8k1k_con666_ctx8_dep4_gen1_dep32_eplb384_mtp3_ccb-NIXL.yaml +++ b/tests/scripts/perf/disaggregated/gb300_deepseek-v4-pro-fp4_8k1k_con666_ctx8_dep4_gen1_dep32_eplb384_mtp3_ccb-NIXL.yaml @@ -1,4 +1,5 @@ metadata: + architecture: DeepseekV4ForCausalLM model_name: deepseek_v4_pro_fp4 precision: fp4 model_dir_name: DeepSeek-V4-Pro diff --git a/tests/scripts/perf/disaggregated/vr200_deepseek-v4-pro-fp4_8k1k_con1229_ctx7_dep4_gen1_dep8_eplb384_mtp3_ccb-NIXL.yaml b/tests/scripts/perf/disaggregated/vr200_deepseek-v4-pro-fp4_8k1k_con1229_ctx7_dep4_gen1_dep8_eplb384_mtp3_ccb-NIXL.yaml index 43dd885b503e..17c285607c2a 100644 --- a/tests/scripts/perf/disaggregated/vr200_deepseek-v4-pro-fp4_8k1k_con1229_ctx7_dep4_gen1_dep8_eplb384_mtp3_ccb-NIXL.yaml +++ b/tests/scripts/perf/disaggregated/vr200_deepseek-v4-pro-fp4_8k1k_con1229_ctx7_dep4_gen1_dep8_eplb384_mtp3_ccb-NIXL.yaml @@ -7,6 +7,7 @@ # first VR200 run is directly comparable to its GB300 counterpart. # Runs on the trtllm-ci VR200-HECATE multi-node chain (partition batch-xdr). metadata: + architecture: DeepseekV4ForCausalLM model_name: deepseek_v4_pro_fp4 precision: fp4 model_dir_name: DeepSeek-V4-Pro diff --git a/tests/scripts/perf/disaggregated/vr200_deepseek-v4-pro-fp4_8k1k_con2_ctx1_dep4_gen5_tep4_eplb0_mtp3_ccb-NIXL.yaml b/tests/scripts/perf/disaggregated/vr200_deepseek-v4-pro-fp4_8k1k_con2_ctx1_dep4_gen5_tep4_eplb0_mtp3_ccb-NIXL.yaml index 70f775078091..43773cc08a17 100644 --- a/tests/scripts/perf/disaggregated/vr200_deepseek-v4-pro-fp4_8k1k_con2_ctx1_dep4_gen5_tep4_eplb0_mtp3_ccb-NIXL.yaml +++ b/tests/scripts/perf/disaggregated/vr200_deepseek-v4-pro-fp4_8k1k_con2_ctx1_dep4_gen5_tep4_eplb0_mtp3_ccb-NIXL.yaml @@ -7,6 +7,7 @@ # first VR200 run is directly comparable to its GB300 counterpart. # Runs on the trtllm-ci VR200-HECATE multi-node chain (partition batch-xdr). metadata: + architecture: DeepseekV4ForCausalLM model_name: deepseek_v4_pro_fp4 precision: fp4 model_dir_name: DeepSeek-V4-Pro diff --git a/tests/unittest/others/test_cache_transceiver_precheck_run.py b/tests/unittest/others/test_cache_transceiver_precheck_run.py index 27182b8ec11f..4069d9500928 100644 --- a/tests/unittest/others/test_cache_transceiver_precheck_run.py +++ b/tests/unittest/others/test_cache_transceiver_precheck_run.py @@ -1322,23 +1322,21 @@ def test_deepseek_v4_auto_selects_kv_cache_manager_v2(self, api, tmp_path): @pytest.mark.parametrize("role", ["ctx", "gen"]) @pytest.mark.parametrize( - "topology", + "config_path", [ - "con1156_ctx2_dep8_gen1_dep8_eplb0_dspark3", - "con1456_ctx3_dep8_gen1_dep16_eplb0_dspark5", + pytest.param(path, id=f"{suite}/{path.stem}") + for suite in ("perf", "perf-sanity") + for path in sorted( + (Path(rp.__file__).parents[2] / suite / "disaggregated").glob("*deepseek-v4*.yaml") + ) ], ) def test_deepseek_v4_declared_architecture_without_checkpoint( - self, api: object, tmp_path: Path, role: str, topology: str + self, api: object, tmp_path: Path, role: str, config_path: Path ) -> None: - """Both failing configurations resolve V2/PYTHON without model weights.""" + """All DeepSeek V4 disagg configurations resolve V2/PYTHON without weights.""" import yaml - config_path = ( - Path(rp.__file__).parent.parent - / "disaggregated" - / (f"gb300_deepseek-v4-pro-dspark_agentx_{topology}_ccb-NIXL.yaml") - ) cfg = yaml.safe_load(config_path.read_text()) assert rp.pcfg.resolve_model_dir(cfg, llm_models_root=str(tmp_path)) is None side = rp.pcfg.side_plan(rp.pcfg.resolve_plan(cfg), role) From c660b5f0efa2a20acf64b7ade4b2ee8e321f23a2 Mon Sep 17 00:00:00 2001 From: Chien-Chun Hung <2679986+chienchunhung@users.noreply.github.com> Date: Mon, 5 Oct 2026 17:17:48 -0700 Subject: [PATCH 4/4] fix: align precheck notices and require regression configs Signed-off-by: Chien-Chun Hung <2679986+chienchunhung@users.noreply.github.com> --- .../cache_transceiver_precheck/run_precheck.py | 13 +++++++------ .../test_cache_transceiver_precheck_run.py | 17 ++++++++++------- 2 files changed, 17 insertions(+), 13 deletions(-) diff --git a/tests/scripts/perf-sanity/cache_transceiver_precheck/run_precheck.py b/tests/scripts/perf-sanity/cache_transceiver_precheck/run_precheck.py index ec6a25a1d511..710d1209e294 100644 --- a/tests/scripts/perf-sanity/cache_transceiver_precheck/run_precheck.py +++ b/tests/scripts/perf-sanity/cache_transceiver_precheck/run_precheck.py @@ -932,6 +932,11 @@ def __init__(self, args, plan, side, comm): self.work_dir = args.work_dir model_dir_resolved = plan.get("_model_dir") is not None kv_shape = plan.get("_kv_shape", {}) + self.synthetic_kv_shape = ( + not model_dir_resolved + or bool(kv_shape.get("simplified")) + or kv_shape.get("source") == "fallback" + ) self.recorder = StatusRecorder( self.work_dir, self.role, @@ -939,11 +944,7 @@ def __init__(self, args, plan, side, comm): self.is_leader, metadata={ "model_dir_resolved": model_dir_resolved, - "synthetic_kv_shape": ( - not model_dir_resolved - or bool(kv_shape.get("simplified")) - or kv_shape.get("source") == "fallback" - ), + "synthetic_kv_shape": self.synthetic_kv_shape, }, ) self.zmq_ctx = None @@ -1011,7 +1012,7 @@ def _resolve_transfer_config(self, kv_shape: dict) -> object: key: kv_shape[key] for key in ("num_layers", "num_kv_heads", "head_dim", "is_mla") }, } - if self.is_leader and (kv_shape.get("simplified") or self.plan.get("_model_dir") is None): + if self.is_leader and self.synthetic_kv_shape: print( f"[precheck {self.role}_{self.server_idx}] synthetic KV pool; " f"model_dir={self.plan.get('_model_dir')!r}, " diff --git a/tests/unittest/others/test_cache_transceiver_precheck_run.py b/tests/unittest/others/test_cache_transceiver_precheck_run.py index 4069d9500928..1b733da4931e 100644 --- a/tests/unittest/others/test_cache_transceiver_precheck_run.py +++ b/tests/unittest/others/test_cache_transceiver_precheck_run.py @@ -45,6 +45,15 @@ KEY = b"k" * 32 +_DEEPSEEK_V4_CONFIGS = [ + (suite, path) + for suite in ("perf", "perf-sanity") + for path in sorted( + (Path(rp.__file__).parents[2] / suite / "disaggregated").glob("*deepseek-v4*.yaml") + ) +] +assert _DEEPSEEK_V4_CONFIGS, "No DeepSeek V4 disaggregated configurations found" + # --------------------------------------------------------------------------- # # rid / seed scheme @@ -1323,13 +1332,7 @@ def test_deepseek_v4_auto_selects_kv_cache_manager_v2(self, api, tmp_path): @pytest.mark.parametrize("role", ["ctx", "gen"]) @pytest.mark.parametrize( "config_path", - [ - pytest.param(path, id=f"{suite}/{path.stem}") - for suite in ("perf", "perf-sanity") - for path in sorted( - (Path(rp.__file__).parents[2] / suite / "disaggregated").glob("*deepseek-v4*.yaml") - ) - ], + [pytest.param(path, id=f"{suite}/{path.stem}") for suite, path in _DEEPSEEK_V4_CONFIGS], ) def test_deepseek_v4_declared_architecture_without_checkpoint( self, api: object, tmp_path: Path, role: str, config_path: Path