From 486699f98cdd0864d2682f04d6ae244cae76c349 Mon Sep 17 00:00:00 2001 From: doxu Date: Tue, 27 Oct 2026 12:03:48 -0700 Subject: [PATCH] test: Phase 1 e2e harness core and TC-01 Brings up a Dynamo KV router, two vLLM workers with KVCR tiers and kvcr_service Guard sidecars on two GPUs, runs a KVCR-disabled reference engine first, and checks TC-01 (cold request commits to the KVCR tier, the repeat is a block-aligned hit) with every answer compared token by token to the reference. Evidence is result.json, JUnit and per-request files. Part 1 of the stacked Phase 1 series; later parts add TC-02..06, disaggregated and two-node topologies, the CI job and the test plan. CI now runs tests/ci next to tests/unit. Signed-off-by: doxu --- .github/workflows/ci.yml | 2 +- .github/workflows/vllm-e2e.yml | 38 ++ tests/ci/test_phase1.py | 386 +++++++++++++++++++ tests/e2e/phase1/__init__.py | 11 + tests/e2e/phase1/cases.py | 289 ++++++++++++++ tests/e2e/phase1/client.py | 167 ++++++++ tests/e2e/phase1/common.py | 202 ++++++++++ tests/e2e/phase1/stack.py | 678 +++++++++++++++++++++++++++++++++ tests/e2e/phase1/telemetry.py | 270 +++++++++++++ tests/e2e/run_phase1.py | 584 ++++++++++++++++++++++++++++ 10 files changed, 2626 insertions(+), 1 deletion(-) create mode 100644 tests/ci/test_phase1.py create mode 100644 tests/e2e/phase1/__init__.py create mode 100644 tests/e2e/phase1/cases.py create mode 100644 tests/e2e/phase1/client.py create mode 100644 tests/e2e/phase1/common.py create mode 100644 tests/e2e/phase1/stack.py create mode 100644 tests/e2e/phase1/telemetry.py create mode 100644 tests/e2e/run_phase1.py diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index e19df6af..86b7b1d6 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -29,4 +29,4 @@ jobs: with: python-version: "3.12" - run: pip install -e . --group dev - - run: python -m pytest tests/unit -q + - run: python -m pytest tests/unit tests/ci -q diff --git a/.github/workflows/vllm-e2e.yml b/.github/workflows/vllm-e2e.yml index b4e62da1..3117e4ef 100644 --- a/.github/workflows/vllm-e2e.yml +++ b/.github/workflows/vllm-e2e.yml @@ -57,3 +57,41 @@ jobs: name: kvcr-vllm-e2e-${{ github.run_attempt }} path: e2e-artifacts/ retention-days: 7 + + phase1: + # KVCR test plan Phase 1 functional gate in Guard mode on the same image and + # runner, after the peer-transfer smoke test. Guard mode needs a runner kernel + # of 6.5 or newer; otherwise the harness reports BLOCKED and the job fails. + # --batch-invariant keeps the bare-engine reference deterministic so token + # comparison is exact. Later parts of the series extend --cases as they add + # cases; TC-05/06 (performance, recovery timing) stay manual until a nightly + # lane exists. + needs: peer-transfer + runs-on: prod-kvcr-tester-amd-gpu-2-v1 + container: + image: ${{ inputs.image }} + timeout-minutes: 60 + env: + HF_HOME: /scratch/huggingface + VLLM_CACHE_ROOT: /scratch/vllm + XDG_CACHE_HOME: /scratch/cache + steps: + - uses: actions/checkout@v4 + - name: Show NVIDIA devices + run: nvidia-smi + - name: Run Phase 1 (TC-01) in Guard mode + shell: bash + env: + MODEL: ${{ inputs.model }} + run: | + set -euo pipefail + python3 -u tests/e2e/run_phase1.py --output-dir artifacts/phase1/run \ + --model "$MODEL" --mode guard --batch-invariant --timeout 2400 \ + --cases TC-01 + - name: Save Phase 1 evidence + if: always() + uses: actions/upload-artifact@v4 + with: + name: kvcr-phase1-${{ github.run_attempt }} + path: artifacts/phase1/ + retention-days: 14 diff --git a/tests/ci/test_phase1.py b/tests/ci/test_phase1.py new file mode 100644 index 00000000..da25945f --- /dev/null +++ b/tests/ci/test_phase1.py @@ -0,0 +1,386 @@ +# SPDX-License-Identifier: Apache-2.0 +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +"""CPU checks of the Phase 1 harness logic; these never claim a GPU pass.""" + +import importlib.util +import json +import sys +import xml.etree.ElementTree as ET +from datetime import datetime, timezone +from pathlib import Path + +import pytest + +E2E = Path(__file__).parents[1] / "e2e" +sys.path.insert(0, str(E2E)) + +from phase1 import cases, common, stack, telemetry # noqa: E402 + +_SPEC = importlib.util.spec_from_file_location( + "kvcr_phase1_runner", E2E / "run_phase1.py" +) +assert _SPEC is not None and _SPEC.loader is not None +runner = importlib.util.module_from_spec(_SPEC) +_SPEC.loader.exec_module(runner) + + +def sample(metrics, stamp="2026-09-25T10:00:05Z"): + values = ", ".join(f"{key}={value}" for key, value in metrics.items()) + return ( + f"{stamp} INFO loggers.log: Engine 000: statistics\n" + f"{stamp} INFO metrics.log: KV Transfer metrics: {values}\n" + ) + + +def at(second): + return datetime(2026, 9, 25, 10, 0, second, tzinfo=timezone.utc) + + +# -- telemetry --------------------------------------------------------------- + + +def test_metric_names_match_vllm_log_form(): + assert telemetry.metric("kvcr_transfer_blocks", "source_write") == ( + "vllm:kvcr_transfer_blocks:('source_write',)" + ) + assert telemetry.duration_count("remote_deliver") == ( + "vllm:kvcr_duration_seconds:('remote_deliver', 'success')_count" + ) + assert telemetry.TIER_READ_BYTES == "vllm:kv_offload_tiering_read_bytes:('1:kvcr',)" + + +def vllm_sample(metrics, stamp): + """A metrics line in vLLM's own log format: second-precision stamp.""" + values = ", ".join(f"{key}={value}" for key, value in metrics.items()) + return ( + f"(MainProcess pid=1) INFO {stamp} [metrics.py:108] " + f"KV Transfer metrics: {values}\n" + ) + + +def test_window_same_second_sample_counts_only_when_flushed_after_the_request(): + # vLLM's logger stamps to the second. A request started at 10:00:10.415; + # a sample flushed at 10:00:10.9 (after Dynamo logged the request at + # 10:00:10.43) carries its delivery and must count; a sample flushed at + # 10:00:10.1, before any request line, belongs to the previous request. + blocks = telemetry.metric(telemetry.TRANSFER_BLOCKS, telemetry.REMOTE_DELIVER) + year = datetime.now(timezone.utc).year + stamp = "09-25 10:00:10" + start = datetime(year, 9, 25, 10, 0, 10, 415000, tzinfo=timezone.utc) + end = start.replace(second=14) + after = ( + vllm_sample({blocks: 5}, "09-25 10:00:00") + + f"{year}-09-25T10:00:10.430000Z INFO handle_payload: request\n" + + vllm_sample({blocks: 22}, stamp) + ) + snapshots, errors = telemetry.parse_log(after) + assert errors == [] + assert ( + telemetry.window(snapshots, errors, start, end).blocks(telemetry.REMOTE_DELIVER) + == 22 + ) + before = ( + vllm_sample({blocks: 5}, "09-25 10:00:00") + + f"{year}-09-25T10:00:09.900000Z INFO handle_payload: previous request\n" + + vllm_sample({blocks: 11}, stamp) + + f"{year}-09-25T10:00:10.430000Z INFO handle_payload: this request\n" + ) + snapshots, errors = telemetry.parse_log(before) + assert ( + telemetry.window(snapshots, errors, start, end).blocks(telemetry.REMOTE_DELIVER) + == 0 + ) + + +def test_parse_log_ignores_naive_or_partial_leading_timestamps(): + # Lines that start with a bare date or a zone-less time must not be taken as + # request stamps: they are naive datetimes and cannot be compared with the + # aware request start (this crashed TC-01 in CI). + blocks = telemetry.metric(telemetry.TRANSFER_BLOCKS, telemetry.REMOTE_DELIVER) + year = datetime.now(timezone.utc).year + text = ( + f"{year}-09-25 some tool banner\n" + f"{year}-09-25T10:00:09 naive stamp without zone\n" + + vllm_sample({blocks: 3}, "09-25 10:00:10") + ) + snapshots, errors = telemetry.parse_log(text) + assert errors == [] and len(snapshots) == 1 + assert snapshots[0].second_precision and snapshots[0].preceded_by is None + start = datetime(year, 9, 25, 10, 0, 10, 415000, tzinfo=timezone.utc) + window = telemetry.window(snapshots, errors, start) + assert window.blocks(telemetry.REMOTE_DELIVER) == 0 + + +def test_window_sums_intervals_and_keeps_latest_gauge(): + blocks = telemetry.metric(telemetry.TRANSFER_BLOCKS, telemetry.REMOTE_DELIVER) + gauge = telemetry.metric(telemetry.STATE, "pinned_keys") + text = ( + sample({blocks: 3, gauge: 7}, "2026-09-25T10:00:01Z") + + sample({blocks: 4, gauge: 2}, "2026-09-25T10:00:06Z") + + sample({blocks: 9, gauge: 0}, "2026-09-25T10:00:30Z") + + "2026-09-25T10:00:31Z INFO metrics.log: KV Transfer metrics: vllm:x=1" + ) + snapshots, errors = telemetry.parse_log(text) + assert errors == [] and len(snapshots) == 3 # the trailing partial line waits + window = telemetry.window(snapshots, errors, at(5), at(10)) + assert window.blocks(telemetry.REMOTE_DELIVER) == 4 + assert window.gauge("pinned_keys") == 2 + assert window.samples == 1 and window.first_after_end is True + everything = telemetry.window(snapshots, errors, at(0)) + assert everything.blocks(telemetry.REMOTE_DELIVER) == 16 + assert everything.gauge("pinned_keys") == 0 + + +def test_window_flags_positive_failed_and_partial_kvcr_metrics(): + failed = telemetry.duration_count(telemetry.REMOTE_DELIVER, "failed") + partial = telemetry.duration_count(telemetry.REMOTE_DELIVER, "partial") + text = sample({failed: 0, partial: 1}) + snapshots, errors = telemetry.parse_log(text) + window = telemetry.window(snapshots, errors, at(0)) + assert [entry["metric"] for entry in window.peer_failures] == [partial] + + +def test_parse_log_reports_incomplete_lines(): + text = "2026-09-25T10:00:05Z INFO KV Transfer metrics: vllm:a=1, vllm:b=oops\n" + snapshots, errors = telemetry.parse_log(text) + assert snapshots == [] and errors and "incomplete" in errors[0] + + +def test_latest_gauge_reads_the_last_sample(tmp_path): + gauge = telemetry.metric(telemetry.STATE, "local_g2_allocated_slots") + log = tmp_path / "worker.log" + log.write_text(sample({gauge: 40}, "2026-09-25T10:00:01Z") + sample({gauge: 41})) + assert telemetry.latest_gauge(log, "local_g2_allocated_slots") == 41 + assert telemetry.latest_gauge(log, "missing") is None + + +# -- token comparison -------------------------------------------------------- + + +def body(prompt, completion, finish="length", extra=None): + payload = { + "choices": [{"text": "x", "finish_reason": finish}], + "usage": {"prompt_tokens_details": {"cached_tokens": 128}}, + "nvext": {"prompt_token_ids": prompt, "completion_token_ids": completion}, + } + payload["nvext"].update(extra or {}) + return payload + + +def test_extract_sample_and_identical_tokens_pass(): + reference = common.extract_sample(body([1, 2, 3], [7, 8])) + observed = common.extract_sample(body([1, 2, 3], [7, 8])) + assert common.compare_tokens(reference, observed) == [] + assert common.cached_tokens(body([1], [2])) == 128 + + +@pytest.mark.parametrize( + "observed, expected", + [ + (body([1, 2, 3], [7, 9]), "diverge at position 1"), + (body([1, 2, 3], [7]), "length 1 != reference 2"), + (body([1, 2, 4], [7, 8]), "prompt token IDs differ"), + (body([1, 2, 3], [7, 8], finish="stop"), "finish reason"), + ], +) +def test_compare_tokens_reports_every_kind_of_mismatch(observed, expected): + reference = common.extract_sample(body([1, 2, 3], [7, 8])) + problems = common.compare_tokens(reference, common.extract_sample(observed)) + assert any(expected in problem for problem in problems) + + +def test_missing_token_ids_is_blocked_not_failed(): + with pytest.raises(common.BlockedError): + common.extract_sample({"choices": [{"text": "x"}], "nvext": {}}) + with pytest.raises(common.BlockedError): + common.extract_sample( + { + "choices": [], + "nvext": {"completion_token_ids": [1], "prompt_token_ids": "x"}, + } + ) + + +# -- prompts ----------------------------------------------------------------- + + +# -- case results, BLOCKED handling, JUnit ----------------------------------- + + +def test_case_result_status_follows_checks(): + result = common.CaseResult("TC-09", "x", "R9") + result.check("a", True) + result.check("b", False, "why") + result.finish() + assert result.status == common.FAIL + empty = common.CaseResult("TC-09", "x", "R9") + empty.finish() + assert empty.status == common.FAIL and "no checks" in empty.error + + +def test_run_case_maps_blocked_and_exceptions_to_statuses(monkeypatch): + def blocked(_ctx): + raise common.BlockedError("needs Linux 6.5") + + def broken(_ctx): + raise RuntimeError("boom") + + monkeypatch.setitem(cases.CASES, "TC-B", blocked) + monkeypatch.setitem(cases.CASES, "TC-X", broken) + assert cases.run_case("TC-B", None).status == common.BLOCKED + failed = cases.run_case("TC-X", None) + assert failed.status == common.FAIL and "boom" in failed.error + + +def test_junit_marks_blocked_as_skipped_and_fail_as_failure(tmp_path): + run = { + "started_at": "2026-09-25T10:00:00+00:00", + "cases": [ + { + "case": "TC-01", + "title": "a", + "status": "PASS", + "checks": [{"name": "n", "passed": True, "detail": "d"}], + "error": None, + }, + { + "case": "TC-02", + "title": "b", + "status": "FAIL", + "checks": [{"name": "n", "passed": False, "detail": "d"}], + "error": "bad", + }, + { + "case": "TC-04", + "title": "c", + "status": "BLOCKED", + "checks": [], + "error": "kernel", + }, + ], + } + common.write_junit(tmp_path, run) + suite = ET.parse(tmp_path / "junit.xml").getroot() + assert suite.attrib["tests"] == "3" + assert suite.attrib["failures"] == "1" and suite.attrib["skipped"] == "1" + kinds = [child.tag for case in suite for child in case] + assert kinds == ["system-out", "failure", "skipped"] + + +# -- stack configuration ----------------------------------------------------- + + +def test_guard_and_nonguard_tier_configs(tmp_path, monkeypatch): + monkeypatch.setattr(stack.Stack, "advertise_host", staticmethod(lambda: "10.0.0.5")) + settings = stack.Settings(model="m", output=tmp_path, mode=stack.GUARD) + guard = stack.Stack(settings, tmp_path / "g") + tier = guard.tier_config(1) + assert tier["kvcr_service_socket_path"].endswith("memory-1.sock") + assert tier["compatibility_digest"] == settings.compatibility_digest + assert tier["control_ports"] == [17772] and "secondary_g2_slots" not in tier + plain = stack.Stack( + stack.Settings(model="m", output=tmp_path, mode=stack.NON_GUARD), tmp_path / "n" + ) + plain_tier = plain.tier_config(0) + assert plain_tier["secondary_g2_slots"] == 256 + assert "kvcr_service_socket_path" not in plain_tier + command = guard.worker_command(0, kvcr=False) + assert "--kv-transfer-config" not in command and "--data-parallel-size" in command + with_kvcr = guard.worker_command(0, kvcr=True) + transfer = json.loads(with_kvcr[with_kvcr.index("--kv-transfer-config") + 1]) + assert transfer["kv_connector_extra_config"]["secondary_tiers"][0]["type"] == "kvcr" + + +def test_kernel_release_parsing(monkeypatch): + monkeypatch.setattr(stack.platform, "release", lambda: "5.15.0-1055-aws") + assert stack.kernel_release() == (5, 15) < stack.MIN_GUARD_KERNEL + monkeypatch.setattr(stack.platform, "release", lambda: "6.14.0-37-generic") + assert stack.kernel_release() >= stack.MIN_GUARD_KERNEL + monkeypatch.setattr(stack.platform, "release", lambda: "weird") + assert stack.kernel_release() == (0, 0) + + +# -- runner arguments -------------------------------------------------------- + + +def test_arguments_validate_gpus_and_cases(tmp_path): + args = runner.arguments(["--output-dir", str(tmp_path), "--cases", "tc-01"]) + assert args.gpus == [0, 1] and args.cases == ["TC-01"] + for bad in ( + ["--gpus", "0"], + ["--gpus", "1,1"], + ["--cases", "TC-99"], + ["--http-port", "8001"], + ["--timeout", "0"], + ): + with pytest.raises(SystemExit): + runner.arguments(["--output-dir", str(tmp_path), *bad]) + + +def test_runner_refuses_non_empty_output(tmp_path): + (tmp_path / "stale").write_text("x") + args = runner.arguments(["--output-dir", str(tmp_path)]) + with pytest.raises(common.ValidationError): + runner.Runner(args) + + +def test_discovery_instance_ids_reads_url_encoded_record_names(tmp_path): + settings = stack.Settings(model="m", output=tmp_path, mode=stack.NON_GUARD) + s = stack.Stack(settings, tmp_path / "s") + s.base_env() + root = Path(s.discovery.name) + (root / "v1" / "instances").mkdir(parents=True) + (root / "v1" / "models").mkdir(parents=True) + for name in ( + "dynamo%2Fbackend%2Fgenerate%2Fc69d07089b72b3c9", + "dynamo%2Fbackend%2Fclear%5Fkv%5Fblocks%2Fc69d07089b72b3c9", + ): + (root / "v1" / "instances" / name).write_text("{}") + (root / "v1" / "models" / "dynamo%2Fbackend%2Fgenerate%2F4ee6").write_text("{}") + ( + root / "v1" / "instances" / "dynamo%2Fbackend%2Fworker%5Fkv%5Fquery%2Fabcd" + ).write_text("{}") + (root / "other").mkdir() + (root / "other" / "ff").write_text("{}") + assert s.discovery_instance_ids() == {str(0xC69D07089B72B3C9), str(0x4EE6)} + assert str(0xC69D07089B72B3C9) == "14311602924402422729" + s.reset_discovery() + + +def test_state_agent_config_only_in_guard_mode(tmp_path, monkeypatch): + monkeypatch.setattr(stack.Stack, "advertise_host", staticmethod(lambda: "10.0.0.5")) + guard = stack.Stack( + stack.Settings(model="m", output=tmp_path, mode=stack.GUARD), tmp_path / "g" + ) + extra = guard.transfer_config(1)["kv_connector_extra_config"] + agent = extra["dynamo_state_agent"] + assert agent["raw_advertise_host"] == "10.0.0.5" + assert agent["cache_owner_ids"] == {"0": stack.Stack.cache_owner_id(1)} + assert stack.Stack.cache_owner_id(1).endswith("/" + "0" * 31 + "1") + plain = stack.Stack( + stack.Settings(model="m", output=tmp_path, mode=stack.NON_GUARD), tmp_path / "n" + ) + assert ( + "dynamo_state_agent" + not in plain.transfer_config(0)["kv_connector_extra_config"] + ) + off = stack.Stack( + stack.Settings(model="m", output=tmp_path, mode=stack.GUARD, state_agent=False), + tmp_path / "o", + ) + assert ( + "dynamo_state_agent" not in off.transfer_config(0)["kv_connector_extra_config"] + ) + + +# -- verdicts ------------------------------------------------------------------- + + +def test_zero_hit_replacement_recovery_fails(): + # A replacement that answers correctly but recomputed the prefix is not a recovery. + assert cases.hit_verdict(0, 64) == (False, "cached_tokens=0 (block 64)") + assert cases.hit_verdict(1472, 64)[0] + assert not cases.hit_verdict(1500, 576)[0], "hybrid hits must be 576-aligned" + assert not cases.hit_verdict(1472, 64, max_blocks=10)[0], ( + "above the shared-prefix bound" + ) diff --git a/tests/e2e/phase1/__init__.py b/tests/e2e/phase1/__init__.py new file mode 100644 index 00000000..aae28740 --- /dev/null +++ b/tests/e2e/phase1/__init__.py @@ -0,0 +1,11 @@ +# SPDX-License-Identifier: Apache-2.0 +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +"""KVCR test plan v7, Phase 1: Guard-mode critical path TC-01 -> TC-04. + +The package runs inside the integration image on one node with two GPUs. It +launches a KVCR-disabled reference engine, records the reference generated +token IDs, then launches a Dynamo KV router and two independent KVCR-enabled +workers (one GPU each) that claim Guard-owned pools from per-worker KVCR +services. Every test case compares generated token IDs with the reference and +requires path evidence from worker and service logs. +""" diff --git a/tests/e2e/phase1/cases.py b/tests/e2e/phase1/cases.py new file mode 100644 index 00000000..7aa8c902 --- /dev/null +++ b/tests/e2e/phase1/cases.py @@ -0,0 +1,289 @@ +# SPDX-License-Identifier: Apache-2.0 +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +"""Phase 1 test cases TC-01 .. TC-04 from the KVCR test plan v7. + +Each case receives the running stack, the reference token samples, and an +evidence directory. A case records checks; the runner derives PASS/FAIL. +BlockedError marks the combination BLOCKED with the missing condition. +""" + +from __future__ import annotations + +import time +from dataclasses import dataclass +from datetime import datetime + +from .client import Client, Response +from .common import ( + BLOCK_SIZE, + BlockedError, + CaseResult, + TokenSample, + ValidationError, + cached_tokens, + compare_tokens, + extract_sample, + utc_now, +) +from .stack import Stack +from .telemetry import ( + LOCAL_DELIVER, + LOCAL_FILL, + TIER_WRITE_BYTES, + Window, + read_window, +) + + +@dataclass(frozen=True) +class Prompts: + """Deterministic prompt texts derived from one recorded run tag.""" + + tag: str + + @staticmethod + def section(tag: str, index: int) -> str: + return f"{tag} section{index} cache routing data" + + def prefix(self, tag: str, start: int, stop: int) -> str: + return " ".join(self.section(tag, i) for i in range(start, stop)) + + @property + def tail(self) -> str: + return " Repeat the first word of this text exactly, then continue the list:" + + @property + def tc01(self) -> str: + return self.prefix(self.tag + "a", 0, 80) + self.tail + + def all(self) -> dict[str, str]: + return {"tc01": self.tc01} + + +def parse_iso(stamp: str) -> datetime: + return datetime.fromisoformat(stamp.replace("Z", "+00:00")) + + +class Context: + """Everything a case needs; also the single place that talks to the stack.""" + + def __init__( + self, + stack: Stack, + client: Client, + prompts: Prompts, + references: dict[str, TokenSample], + *, + index_wait: float, + telemetry_timeout: float, + hit_block_tokens: int = BLOCK_SIZE, + ) -> None: + self.stack = stack + self.client = client + self.prompts = prompts + self.references = references + self.index_wait = index_wait + self.telemetry_timeout = telemetry_timeout + # Tokens per prefix-cache hit block. 64 (vLLM --block-size) for dense + # models; hybrid models (Qwen3.5) raise the attention block to 576 so + # cached_tokens is a multiple of that. + self.hit_block_tokens = hit_block_tokens + + # -- request helpers --------------------------------------------------- + + def request(self, stage: str, prompt: str, worker_index: int) -> Response: + self.stack.check_children() + worker_id = self.stack.worker_ids[worker_index] + del worker_id + return self.client.completion(stage, prompt, **self.stack.pin(worker_index)) + + def require_ok( + self, result: CaseResult, response: Response, worker_index: int + ) -> bool: + ok = result.check( + f"{response.stage}: HTTP 200 with JSON body", + response.ok, + response.error or f"status {response.http_status}", + ) + if not ok: + return False + worker = self.stack.identity(response.body or {}) + return result.check( + f"{response.stage}: served by worker {worker_index}", + worker == self.stack.worker_ids[worker_index], + f"{self.stack.served_role}_worker_id={worker}", + ) + + def compare( + self, result: CaseResult, response: Response, reference_key: str + ) -> bool: + if not response.ok or response.body is None: + return result.check( + f"{response.stage}: tokens equal reference", False, "no body" + ) + sample = extract_sample(response.body) + problems = compare_tokens(self.references[reference_key], sample) + return result.check( + f"{response.stage}: tokens equal reference ({reference_key})", + not problems, + "; ".join(problems) + or f"{sample.length} tokens, finish={sample.finish_reason}", + ) + + def cached(self, response: Response) -> int: + value = cached_tokens(response.body or {}) + if value is None: + raise ValidationError(f"{response.stage}: response lacks cached_tokens") + return value + + def check_cold(self, result: CaseResult, response: Response) -> None: + cached = self.cached(response) + result.check( + f"{response.stage}: cold request has 0 cached tokens", + cached == 0, + str(cached), + ) + + def check_hit( + self, result: CaseResult, response: Response, *, max_blocks: int | None = None + ) -> int: + cached = self.cached(response) + block = self.hit_block_tokens + aligned, detail = hit_verdict(cached, block, max_blocks) + result.check( + f"{response.stage}: cache hit is positive and block-aligned", + aligned, + detail, + ) + return cached // block + + # -- telemetry helpers ------------------------------------------------- + + def wait_window( + self, + worker_index: int, + start: str, + predicate, + *, + tag: str = "", + ) -> Window: + """Poll a worker's log until the summed window satisfies predicate. + + The engine emits one interval sample per logging period; the sample + that covers a request can arrive several seconds after the response. + """ + started = parse_iso(start) + deadline = time.monotonic() + self.telemetry_timeout + log = self.stack.workers[worker_index].log_path + result = read_window(log, started) + while time.monotonic() < deadline: + result = read_window(log, started) + if predicate(result): + break + time.sleep(1) + return result + + def settle(self, worker_index: int, start: str) -> Window: + """Wait one extra interval so the window includes the request's sample.""" + started = parse_iso(start) + log = self.stack.workers[worker_index].log_path + deadline = time.monotonic() + self.telemetry_timeout + while time.monotonic() < deadline: + current = read_window(log, started) + if current.samples >= 2: + return current + time.sleep(1) + return read_window(log, started) + + def local_commit_checks( + self, result: CaseResult, worker_index: int, start: str, label: str + ): + """The cold request's blocks were stored and committed into the KVCR tier.""" + window = self.wait_window( + worker_index, start, lambda w: w.blocks(LOCAL_FILL) > 0 + ) + fills = window.blocks(LOCAL_FILL) + result.check( + f"{label}: KVCR local_fill committed blocks after cold request", + fills > 0 and window.successes(LOCAL_FILL) >= 1, + f"local_fill blocks={fills}, successes={window.successes(LOCAL_FILL)}, " + f"tier write bytes={window.get(TIER_WRITE_BYTES)}", + ) + result.check( + f"{label}: no positive KVCR failed/partial metrics", + not window.peer_failures, + str(window.peer_failures), + ) + result.measurements[label] = window.totals + return window + + +def hit_verdict( + cached: int, block: int, max_blocks: int | None = None +) -> tuple[bool, str]: + """A served prefix counts as a hit only if cached_tokens is a positive + multiple of the hit block (and within the bound when one is given). A + correct answer with zero cached tokens is a recomputation, not a recovery. + """ + aligned = cached > 0 and cached % block == 0 + detail = f"cached_tokens={cached} (block {block})" + if max_blocks is not None: + detail += f", upper bound {max_blocks * block}" + aligned = aligned and cached <= max_blocks * block + return aligned, detail + + +# -- cases ------------------------------------------------------------------- + + +def tc01(ctx: Context) -> CaseResult: + result = CaseResult("TC-01", "Integration generation correctness", "R1") + prompt = ctx.prompts.tc01 + cold = ctx.request("tc01-cold-w0", prompt, 0) + result.evidence += [cold.request_file, cold.response_file] + if not ctx.require_ok(result, cold, 0): + result.finish() + return result + ctx.compare(result, cold, "tc01") + ctx.check_cold(result, cold) + ctx.local_commit_checks(result, 0, cold.started_at, "tc01 store->commit") + time.sleep(ctx.index_wait) + reuse = ctx.request("tc01-reuse-w0", prompt, 0) + result.evidence += [reuse.request_file, reuse.response_file] + if ctx.require_ok(result, reuse, 0): + ctx.compare(result, reuse, "tc01") + ctx.check_hit(result, reuse) + window = ctx.settle(0, reuse.started_at) + result.check( + "tc01-reuse-w0: no positive KVCR failed/partial metrics", + not window.peer_failures, + str(window.peer_failures), + ) + result.measurements["reuse_window"] = window.totals + result.measurements["reuse_kvcr_local_deliver_blocks"] = window.blocks( + LOCAL_DELIVER + ) + result.finish() + return result + + +CASES = {"TC-01": tc01} + + +def run_case(name: str, ctx: Context) -> CaseResult: + started = utc_now() + try: + result = CASES[name](ctx) + except BlockedError as blocked: + result = CaseResult(name, name, "") + result.status = "BLOCKED" + result.error = str(blocked) + result.started_at = started + result.finish() + except Exception as error: # noqa: BLE001 - every failure becomes evidence + result = CaseResult(name, name, "") + result.status = "FAIL" + result.error = f"{type(error).__name__}: {error}" + result.started_at = started + result.finish() + return result diff --git a/tests/e2e/phase1/client.py b/tests/e2e/phase1/client.py new file mode 100644 index 00000000..164daf01 --- /dev/null +++ b/tests/e2e/phase1/client.py @@ -0,0 +1,167 @@ +# SPDX-License-Identifier: Apache-2.0 +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +"""OpenAI-compatible completions client that records every request and reply.""" + +from __future__ import annotations + +import json +import time +from dataclasses import dataclass, field +from pathlib import Path +from typing import Any +from urllib.error import HTTPError +from urllib.request import Request, urlopen + +from .common import save_json, utc_now + +EXTRA_FIELDS = [ + "worker_id", + "prompt_token_ids", + "completion_token_ids", + "detailed_finish_reason", +] + + +@dataclass +class Response: + stage: str + started_at: str + finished_at: str + latency_seconds: float + http_status: int | None + body: dict[str, Any] | None + error: str | None + request_file: str + response_file: str + headers: dict[str, str] = field(default_factory=dict) + + @property + def ok(self) -> bool: + return ( + self.http_status == 200 and isinstance(self.body, dict) and not self.error + ) + + +def _nvext(chunk: dict[str, Any]) -> dict[str, Any]: + top = chunk.get("nvext") + if isinstance(top, dict): + return top + choices = chunk.get("choices") or [] + if choices and isinstance(choices[0], dict): + inner = choices[0].get("nvext") + if isinstance(inner, dict): + return inner + return {} + + +class Client: + def __init__( + self, + base_url: str, + model: str, + evidence: Path, + *, + max_tokens: int, + request_timeout: float, + ) -> None: + self.base_url = base_url.rstrip("/") + self.model = model + self.evidence = evidence + self.max_tokens = max_tokens + self.request_timeout = request_timeout + self.log: list[dict[str, Any]] = [] + + def models(self, timeout: float) -> dict[str, Any]: + with urlopen(self.base_url + "/v1/models", timeout=timeout) as response: + return json.load(response) + + def completion( + self, + stage: str, + prompt: str, + *, + worker: str | None = None, + rank: int | None = None, + max_tokens: int | None = None, + extra_fields: list[str] | None = None, + ) -> Response: + headers = {"Content-Type": "application/json"} + if worker is not None: + headers["x-dynamo-worker-instance-id"] = str(worker) + if rank is not None: + headers["x-dynamo-dp-rank"] = str(rank) + body = { + "model": self.model, + "prompt": prompt, + "max_tokens": self.max_tokens if max_tokens is None else max_tokens, + "temperature": 0, + "seed": 0, + "nvext": {"extra_fields": extra_fields or EXTRA_FIELDS}, + } + started_at = utc_now() + request_file = f"{stage}.request.json" + response_file = f"{stage}.response.json" + save_json( + self.evidence / request_file, + { + "started_at": started_at, + "url": self.base_url + "/v1/completions", + "headers": headers, + "body": body, + }, + ) + started = time.monotonic() + status: int | None = None + raw = b"" + parsed: dict[str, Any] | None = None + error: str | None = None + response_headers: dict[str, str] = {} + try: + request = Request( + self.base_url + "/v1/completions", + headers=headers, + data=json.dumps(body).encode(), + method="POST", + ) + try: + http = urlopen(request, timeout=self.request_timeout) + except HTTPError as failure: + http = failure + with http: + status = http.status + response_headers = dict(http.headers.items()) + raw = http.read() + parsed = json.loads(raw) + if not isinstance(parsed, dict): + parsed = None + error = "response body is not a JSON object" + except Exception as failure: # noqa: BLE001 - evidence keeps every failure + error = f"{type(failure).__name__}: {failure}" + result = Response( + stage=stage, + started_at=started_at, + finished_at=utc_now(), + latency_seconds=time.monotonic() - started, + http_status=status, + body=parsed, + error=error, + request_file=request_file, + response_file=response_file, + headers=response_headers, + ) + record = { + **result.__dict__, + "body_text": None if parsed is not None else raw.decode(errors="replace"), + } + save_json(self.evidence / response_file, record) + self.log.append( + { + "stage": stage, + "started_at": started_at, + "finished_at": result.finished_at, + "http_status": status, + "request_file": request_file, + "response_file": response_file, + } + ) + return result diff --git a/tests/e2e/phase1/common.py b/tests/e2e/phase1/common.py new file mode 100644 index 00000000..fa7d8d94 --- /dev/null +++ b/tests/e2e/phase1/common.py @@ -0,0 +1,202 @@ +# SPDX-License-Identifier: Apache-2.0 +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +"""Shared result types, token comparison, and evidence writers.""" + +from __future__ import annotations + +import json +import re +import xml.etree.ElementTree as ET +from dataclasses import asdict, dataclass, field +from datetime import datetime, timezone +from pathlib import Path +from typing import Any + +BLOCK_SIZE = 64 +PASS = "PASS" +FAIL = "FAIL" +BLOCKED = "BLOCKED" +ANSI = re.compile(r"\x1b\[[0-?]*[ -/]*[@-~]") + + +class ValidationError(RuntimeError): + """Observed behaviour or evidence does not satisfy the test plan.""" + + +class BlockedError(RuntimeError): + """The combination cannot run here; record the missing condition, not FAIL.""" + + +def utc_now() -> str: + return datetime.now(timezone.utc).isoformat(timespec="milliseconds") + + +def save_json(path: Path, value: Any) -> None: + temporary = path.with_name(path.name + ".tmp") + temporary.write_text(json.dumps(value, indent=2, sort_keys=True) + "\n") + temporary.replace(path) + + +@dataclass(frozen=True) +class TokenSample: + """Generated token IDs and finish reason captured from one response.""" + + prompt_token_ids: tuple[int, ...] + completion_token_ids: tuple[int, ...] + finish_reason: str | None + + @property + def length(self) -> int: + return len(self.completion_token_ids) + + +def extract_sample(body: dict[str, Any]) -> TokenSample: + """Read token IDs from a Dynamo completions response. + + The plan requires token IDs captured from the engine layer or, through the + frontend, a response field carrying them. Absence is BLOCKED, not a + comparison failure. + """ + nvext = body.get("nvext") or {} + completion = nvext.get("completion_token_ids") + prompt = nvext.get("prompt_token_ids") + if not isinstance(completion, list) or not all( + type(token) is int for token in completion + ): + raise BlockedError( + "response carries no nvext.completion_token_ids; token capture through " + "the Dynamo frontend is unavailable on this build" + ) + if not isinstance(prompt, list) or not all(type(token) is int for token in prompt): + raise BlockedError("response carries no nvext.prompt_token_ids") + choices = body.get("choices") or [] + finish = choices[0].get("finish_reason") if choices else None + return TokenSample(tuple(prompt), tuple(completion), finish) + + +def compare_tokens(reference: TokenSample, observed: TokenSample) -> list[str]: + """Return the list of mismatches; empty means identical. + + Exactly two outcomes exist in the plan: identical passes, everything else + is an error with no distinction by cause. The messages are diagnostics. + """ + problems = [] + if observed.prompt_token_ids != reference.prompt_token_ids: + problems.append("effective prompt token IDs differ from the reference") + if observed.length != reference.length: + problems.append( + f"generated length {observed.length} != reference {reference.length}" + ) + if observed.completion_token_ids != reference.completion_token_ids: + first = next( + ( + index + for index, (left, right) in enumerate( + zip(reference.completion_token_ids, observed.completion_token_ids) + ) + if left != right + ), + min(reference.length, observed.length), + ) + problems.append(f"generated token IDs diverge at position {first}") + if observed.finish_reason != reference.finish_reason: + problems.append( + f"finish reason {observed.finish_reason!r} != " + f"reference {reference.finish_reason!r}" + ) + return problems + + +def cached_tokens(body: dict[str, Any]) -> int | None: + details = (body.get("usage") or {}).get("prompt_tokens_details") or {} + value = details.get("cached_tokens") + return value if type(value) is int else None + + +def worker_identity( + body: dict[str, Any], role: str = "decode" +) -> tuple[str | None, int | None]: + identity = (body.get("nvext") or {}).get("worker_id") or {} + worker = identity.get(f"{role}_worker_id") + rank = identity.get("decode_dp_rank") + return ( + str(worker) if type(worker) in (str, int) and str(worker) else None, + rank if type(rank) is int else None, + ) + + +@dataclass +class Check: + name: str + passed: bool + detail: str = "" + + +@dataclass +class CaseResult: + case: str + title: str + requirement: str + status: str = "RUNNING" + checks: list[Check] = field(default_factory=list) + evidence: list[str] = field(default_factory=list) + error: str | None = None + started_at: str = field(default_factory=utc_now) + finished_at: str | None = None + measurements: dict[str, Any] = field(default_factory=dict) + + def check(self, name: str, passed: bool, detail: str = "") -> bool: + self.checks.append(Check(name, bool(passed), detail)) + return bool(passed) + + def finish(self) -> None: + self.finished_at = utc_now() + if self.status == "RUNNING": + self.status = PASS if all(check.passed for check in self.checks) else FAIL + if not self.checks: + self.status = FAIL + self.error = self.error or "no checks were recorded" + + def to_dict(self) -> dict[str, Any]: + return asdict(self) + + +def write_junit(output: Path, run: dict[str, Any]) -> None: + cases = run.get("cases", []) + suite = ET.Element( + "testsuite", + name="kvcr-phase1", + tests=str(len(cases)), + failures=str(sum(case["status"] == FAIL for case in cases)), + skipped=str(sum(case["status"] == BLOCKED for case in cases)), + errors="0", + timestamp=run.get("started_at", ""), + ) + for case in cases: + element = ET.SubElement( + suite, + "testcase", + classname="kvcr.phase1", + name=f"{case['case']} {case['title']}", + ) + lines = [ + f"[{'ok' if check['passed'] else 'FAIL'}] " + f"{check['name']}: {check['detail']}" + for check in case["checks"] + ] + if case.get("error"): + lines.append(f"error: {case['error']}") + text = "\n".join(lines) + if case["status"] == FAIL: + ET.SubElement( + element, "failure", message=case.get("error") or "FAIL" + ).text = text + elif case["status"] == BLOCKED: + ET.SubElement( + element, "skipped", message=case.get("error") or "BLOCKED" + ).text = text + else: + ET.SubElement(element, "system-out").text = text + ET.ElementTree(suite).write( + output / "junit.xml", encoding="unicode", xml_declaration=True + ) diff --git a/tests/e2e/phase1/stack.py b/tests/e2e/phase1/stack.py new file mode 100644 index 00000000..544ab963 --- /dev/null +++ b/tests/e2e/phase1/stack.py @@ -0,0 +1,678 @@ +# SPDX-License-Identifier: Apache-2.0 +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +"""Process management for the Phase 1 stack. + +Topology (one node, two GPUs, aggregated, TP=1): + + frontend (dynamo.frontend --router-mode kv) + worker[0] GPU 0 DP=1 KVCR tier -> kvcr_service[0] (Guard 0) + worker[1] GPU 1 DP=1 KVCR tier -> kvcr_service[1] (Guard 0) + +Every worker is its own process group so a test can kill one engine without +touching its KVCR service, which must outlive the engine for TC-04. +""" + +from __future__ import annotations + +import ctypes.util +import importlib.util +import json +import os +import platform +import re +import signal +import socket +import subprocess +import sys +import tempfile +import time +import uuid +from dataclasses import dataclass, field +from pathlib import Path +from typing import Any, Callable +from urllib.parse import unquote + +from .client import Client +from .common import ( + BLOCK_SIZE, + BlockedError, + ValidationError, + save_json, + utc_now, + worker_identity, +) + +GUARD = "guard" +NON_GUARD = "nonguard" +DISABLED = "disabled" +MODES = (GUARD, NON_GUARD) +MIN_GUARD_KERNEL = (6, 5) +GUARD_PROMOTED = "KVCR_EVENT guard_promoted" +STATE_AGENT_READY = "KV state-agent host started" +CACHE_OWNER_DOMAIN = ( + "01010101010101010101010101010101:02020202020202020202020202020202/0303030303030303" +) +RECOVERY_DISABLED = "KVCR pool recovery disabled" + + +def kernel_release() -> tuple[int, int]: + match = re.match(r"(\d+)\.(\d+)", platform.release()) + if match is None: + return (0, 0) + return int(match.group(1)), int(match.group(2)) + + +def check_port_available(port: int) -> None: + """Reject active listeners; allow reuse after a prior service exited.""" + try: + connection = socket.create_connection(("127.0.0.1", port), timeout=0.2) + except OSError: + pass + else: + connection.close() + raise ValidationError(f"required TCP port {port} has an active listener") + with socket.socket() as listener: + listener.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) + try: + listener.bind(("0.0.0.0", port)) + listener.listen(1) + except OSError as error: + raise ValidationError(f"required TCP port {port} is unavailable: {error}") + + +def kvcr_service_command() -> list[str]: + """Interpreter command that starts the KVCR service. + + KVCR 0.2.0 moved the service to ``kvcr.service.server`` (also installed as + the ``kvcr-service`` console script); earlier releases expose it as + ``kvcr.kvcr_service``. Pick by what the installed package provides so the + harness runs against both. + """ + try: + new_layout = importlib.util.find_spec("kvcr.service") is not None + except ModuleNotFoundError: + # kvcr itself is missing; the service process reports that clearly. + new_layout = False + module = "kvcr.service.server" if new_layout else "kvcr.kvcr_service" + return [sys.executable, "-m", module] + + +@dataclass +class Settings: + model: str + output: Path + mode: str = GUARD + # Two decode workers: worker 0 is the KVCR source, worker 1 the target. + gpus: tuple[int, ...] = (0, 1) + http_port: int = 8000 + reference_http_port: int = 8001 + control_ports: tuple[int, ...] = (17771, 17772) + event_ports: tuple[int, ...] = (20081, 20082) + system_ports: tuple[int, ...] = (9091, 9092) + pool_gb: int = 2 + g2_slots: int = 256 + cpu_bytes_to_use: int = 2_000_000_000 + max_model_len: int = 32000 + gpu_memory_utilization: float = 0.8 + compatibility_digest: str = "" + batch_invariant: bool = False + state_agent: bool = True + # KVCR's quick-start disables vLLM's hybrid KV cache manager. Models with + # non-attention layers (Qwen3.5 GDN, Mamba hybrids) cannot be unified into + # one KV group, so they need the manager kept on. + disable_hybrid_kv_cache_manager: bool = True + # Eager mode keeps the stack simple, but GDN/Mamba layers without CUDA + # graphs decode very slowly (Qwen3.5 reference phase took ~40 min). + enforce_eager: bool = True + readiness_timeout: float = 600 + request_timeout: float = 120 + extra_worker_args: list[str] = field(default_factory=list) + # Extra environment for the Dynamo frontend only (e.g. DYN_LOG for router + # debugging). + frontend_env: dict[str, str] = field(default_factory=dict) + + def __post_init__(self) -> None: + if not self.compatibility_digest: + self.compatibility_digest = ( + f"kvcr-phase1-{self.model}-{uuid.uuid4().hex[:8]}" + ) + + +class Managed: + """A child process in its own session with a log file.""" + + def __init__(self, name: str, command: list[str], env: dict[str, str], log: Path): + self.name = name + self.command = command + self.log_path = log + self.started_at = utc_now() + self.stopped = False + self._log = log.open("ab") + try: + self.process = subprocess.Popen( + command, + stdout=self._log, + stderr=subprocess.STDOUT, + env=env, + start_new_session=True, + ) + except BaseException: + self._log.close() + raise + + @property + def pid(self) -> int: + return self.process.pid + + def alive(self) -> bool: + return self.process.poll() is None + + def signal_group(self, number: int) -> bool: + try: + os.killpg(self.process.pid, number) + return True + except ProcessLookupError: + return False + + def stop(self, grace: float = 20) -> dict[str, Any]: + outcome: dict[str, Any] = {"name": self.name, "pid": self.process.pid} + self.stopped = True + self.signal_group(signal.SIGTERM) + try: + self.process.wait(timeout=grace) + except subprocess.TimeoutExpired: + outcome["sigkill"] = True + # Descendants may survive their leader; sweep the whole group. + self.signal_group(signal.SIGKILL) + try: + self.process.wait(timeout=10) + outcome["exit_code"] = self.process.returncode + except subprocess.TimeoutExpired: + outcome["error"] = "process did not exit" + self._log.close() + return outcome + + def log_text(self) -> str: + return self.log_path.read_text(errors="replace") + + +class Stack: + def __init__(self, settings: Settings, evidence: Path) -> None: + self.settings = settings + self.evidence = evidence + self.evidence.mkdir(parents=True, exist_ok=True) + self.children: list[Managed] = [] + self.discovery: tempfile.TemporaryDirectory | None = None + self.services: dict[int, Managed] = {} + self.state_agent: Managed | None = None + self.workers: dict[int, Managed] = {} + self.worker_ids: dict[int, str] = {} + self.retired_ids: set[str] = set() + self.frontend: Managed | None = None + self.kvcr_run_dir = Path(tempfile.mkdtemp(prefix="phase1-run-")) + self.pool_dir = Path("/dev/shm/phase1-pool-" + uuid.uuid4().hex[:8]) + self.deadline_check: Callable[[], None] = lambda: None + + # -- preflight --------------------------------------------------------- + + def preflight(self) -> dict[str, Any]: + import torch + + shm = os.statvfs("/dev/shm") + data: dict[str, Any] = { + "kernel": platform.release(), + "cuda_available": torch.cuda.is_available(), + "device_count": torch.cuda.device_count(), + "shm_available_bytes": shm.f_bavail * shm.f_frsize, + "libatomic": ctypes.util.find_library("atomic"), + "mode": self.settings.mode, + "devices": [], + } + save_json(self.evidence / "preflight.json", data) + needed = max(self.settings.gpus) + 1 + if not data["cuda_available"] or data["device_count"] < needed: + raise BlockedError(f"Phase 1 needs {needed} usable CUDA GPU(s) on one node") + for index in sorted(set(self.settings.gpus)): + value = torch.ones(16, device=f"cuda:{index}").sum().item() + data["devices"].append( + { + "index": index, + "name": torch.cuda.get_device_name(index), + "probe": value, + } + ) + if value != 16: + raise ValidationError(f"CUDA compute probe failed on GPU {index}") + if self.settings.mode == GUARD: + if kernel_release() < MIN_GUARD_KERNEL: + raise BlockedError( + "Guard mode needs Linux " + f"{MIN_GUARD_KERNEL[0]}.{MIN_GUARD_KERNEL[1]}+ " + f"(SO_PEERPIDFD); this node runs {platform.release()}" + ) + if data["libatomic"] is None: + raise BlockedError("Guard mode needs the system libatomic runtime") + needed = 2 * (self.settings.pool_gb * 1024**3 + 128 * 1024**2) + if data["shm_available_bytes"] < needed: + raise BlockedError( + f"/dev/shm has {data['shm_available_bytes']} bytes free; two Guard " + f"pools need at least {needed}" + ) + for port in ( + self.settings.http_port, + self.settings.reference_http_port, + *self.settings.control_ports, + *self.settings.event_ports, + *self.settings.system_ports, + ): + check_port_available(port) + save_json(self.evidence / "preflight.json", data) + return data + + # -- environment ------------------------------------------------------- + + def base_env(self) -> dict[str, str]: + if self.discovery is None: + self.discovery = tempfile.TemporaryDirectory(prefix="phase1-disc-") + env = os.environ.copy() + env.pop("NATS_SERVER", None) + env.update( + DYN_REQUEST_PLANE="tcp", + DYN_EVENT_PLANE="zmq", + PYTHONHASHSEED="0", + VLLM_KV_EVENTS_USE_INT_BLOCK_HASHES="1", + ) + env["DYN_DISCOVERY_BACKEND"] = "file" + env["DYN_FILE_KV"] = self.discovery.name + if self.settings.batch_invariant: + env["VLLM_BATCH_INVARIANT"] = "1" + return env + + def reset_discovery(self) -> None: + if self.discovery is not None: + self.discovery.cleanup() + self.discovery = None + + # -- configuration ----------------------------------------------------- + + def tier_config(self, index: int) -> dict[str, Any]: + tier: dict[str, Any] = { + "type": "kvcr", + "router_capabilities": ["router_hint"], + "control_host": "0.0.0.0", + "control_ports": [self.settings.control_ports[index]], + "control_advertise_host": "127.0.0.1", + "operation_timeout_ms": 5000, + "abandon_timeout_ms": 10000, + "enable_telemetry": True, + } + if self.settings.mode == GUARD: + tier["kvcr_service_socket_path"] = str(self.service_socket(index)) + tier["compatibility_digest"] = self.settings.compatibility_digest + else: + tier["secondary_g2_slots"] = self.settings.g2_slots + return tier + + @staticmethod + def cache_owner_id(slot: int) -> str: + """Stable per-slot CacheOwnerId; a replacement worker reuses its slot.""" + return f"{CACHE_OWNER_DOMAIN}/{slot:032x}" + + @staticmethod + def advertise_host() -> str: + """A non-loopback address of this host for the state agent's raw ZMQ link.""" + host = socket.gethostbyname(socket.gethostname()) + if host.startswith("127."): + with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as probe: + probe.connect(("10.255.255.255", 1)) + host = probe.getsockname()[0] + return host + + def uses_state_agent(self) -> bool: + return self.settings.mode == GUARD and self.settings.state_agent + + def transfer_config(self, index: int) -> dict[str, Any]: + extra: dict[str, Any] = { + "spec_name": "TieringOffloadingSpec", + "cpu_bytes_to_use": self.settings.cpu_bytes_to_use, + "self_describing_kv_events": True, + "secondary_tiers": [self.tier_config(index)], + } + if self.uses_state_agent(): + extra["dynamo_state_agent"] = { + "raw_advertise_host": self.advertise_host(), + "cache_owner_ids": {"0": self.cache_owner_id(index)}, + } + return { + "kv_connector": "OffloadingConnector", + "kv_role": "kv_both", + "kv_connector_extra_config": extra, + } + + @property + def served_role(self) -> str: + """Which worker id identifies the KVCR-relevant worker of a response.""" + return "decode" + + def identity(self, body: dict[str, Any]) -> str | None: + return worker_identity(body, self.served_role)[0] + + def served(self, stream: Any) -> str | None: + return stream.worker + + def pin(self, index: int) -> dict[str, str | None]: + """Request kwargs that route a request through worker[index].""" + return {"worker": self.worker_ids.get(index)} + + def events_config(self, index: int) -> dict[str, Any]: + return { + "publisher": "zmq", + "topic": "kv-events", + "endpoint": f"tcp://*:{self.settings.event_ports[index]}", + "enable_kv_cache_events": True, + } + + def worker_command(self, index: int, *, kvcr: bool) -> list[str]: + command = [ + sys.executable, + "-m", + "dynamo.vllm", + "--model", + self.settings.model, + "--tensor-parallel-size", + "1", + "--data-parallel-size", + "1", + "--block-size", + str(BLOCK_SIZE), + "--max-model-len", + str(self.settings.max_model_len), + "--enable-prefix-caching", + *( + ["--disable-hybrid-kv-cache-manager"] + if self.settings.disable_hybrid_kv_cache_manager + else [] + ), + *(["--enforce-eager"] if self.settings.enforce_eager else []), + "--gpu-memory-utilization", + str(self.settings.gpu_memory_utilization), + *self.settings.extra_worker_args, + ] + if kvcr: + command += [ + "--kv-transfer-config", + json.dumps(self.transfer_config(index)), + "--kv-events-config", + json.dumps(self.events_config(index)), + ] + return command + + def service_socket(self, index: int) -> Path: + return self.kvcr_run_dir / f"memory-{index}.sock" + + # -- launching --------------------------------------------------------- + + def _spawn(self, name: str, command: list[str], env: dict[str, str]) -> Managed: + log = self.evidence / f"{name}.log" + child = Managed(name, command, env, log) + self.children.append(child) + save_json( + self.evidence / f"{name}.launch.json", + { + "command": command, + "pid": child.pid, + "started_at": child.started_at, + "environment_overrides": { + key: env[key] + for key in sorted(env) + if key.startswith( + ("DYN_", "VLLM_", "CUDA_", "KVCR_", "PYTHONHASHSEED") + ) + }, + }, + ) + return child + + def start_frontend(self, port: int, name: str = "frontend") -> Managed: + command = [ + sys.executable, + "-m", + "dynamo.frontend", + "--router-mode", + "kv", + "--http-port", + str(port), + ] + env = self.base_env() + env.update(self.settings.frontend_env) + self.frontend = self._spawn(name, command, env) + return self.frontend + + def start_state_agent(self) -> Managed: + """Run Dynamo's KV state-agent host so a dead engine's cache stays routable. + + Without it the router drops a vanished worker's blocks and never hints + at its Guard. One host serves both worker slots. + """ + command = [ + sys.executable, + "-m", + "dynamo.kv_state_agent", + "--max-slots", + "2", + ] + env = self.base_env() + env["DYN_SYSTEM_PORT"] = str(max(self.settings.system_ports) + 1) + agent = self._spawn("kv-state-agent", command, env) + self.state_agent = agent + self.wait_log(agent, STATE_AGENT_READY, 120) + return agent + + def start_service(self, index: int) -> Managed: + if self.settings.mode != GUARD: + raise ValidationError("KVCR services are only launched in Guard mode") + pool_dir = self.pool_dir / str(index) + pool_dir.mkdir(parents=True, exist_ok=True) + command = [ + *kvcr_service_command(), + "--socket-path", + str(self.service_socket(index)), + "--pool-dir", + str(pool_dir), + "--guard-count", + "1", + "--pool-sizes-gb", + str(self.settings.pool_gb), + "--compatibility-digest", + self.settings.compatibility_digest, + "--log-level", + "DEBUG", + ] + service = self._spawn(f"kvcr-service-{index}", command, self.base_env()) + self.services[index] = service + deadline = time.monotonic() + 60 + while time.monotonic() < deadline: + if not service.alive(): + raise ValidationError( + f"kvcr_service {index} exited early: {service.process.returncode}" + ) + try: + probe = socket.socket(socket.AF_UNIX) + probe.settimeout(1) + probe.connect(str(self.service_socket(index))) + probe.close() + return service + except OSError: + time.sleep(0.5) + raise ValidationError(f"kvcr_service {index} socket did not become ready") + + def start_worker(self, index: int, *, kvcr: bool = True, tag: str = "") -> Managed: + env = self.base_env() + env["CUDA_VISIBLE_DEVICES"] = str(self.settings.gpus[index]) + env["VLLM_LOGGING_LEVEL"] = "DEBUG" + env["DYN_SYSTEM_PORT"] = str(self.settings.system_ports[index]) + name = f"worker-{index}{tag}" + save_json( + self.evidence / f"{name}.kv-transfer-config.json", + self.transfer_config(index) if kvcr else None, + ) + worker = self._spawn(name, self.worker_command(index, kvcr=kvcr), env) + self.workers[index] = worker + previous = self.worker_ids.pop(index, None) + if previous is not None: + self.retired_ids.add(previous) + return worker + + # -- readiness --------------------------------------------------------- + + def check_children(self, *names: str) -> None: + for child in self.children: + if names and child.name not in names: + continue + if child in self.workers.values() or child is self.frontend: + if not child.alive(): + raise ValidationError( + f"{child.name} exited early with status " + f"{child.process.returncode}" + ) + elif ( + child in list(self.services.values()) or child is self.state_agent + ) and not child.alive(): + raise ValidationError( + f"{child.name} exited with status {child.process.returncode}" + ) + + def wait_model(self, client: Client) -> dict[str, Any]: + deadline = time.monotonic() + self.settings.readiness_timeout + last = "model not registered" + while time.monotonic() < deadline: + self.deadline_check() + self.check_children() + try: + data = client.models(timeout=5) + except Exception as error: # noqa: BLE001 - readiness polling + last = str(error) + else: + if any( + m.get("id") == self.settings.model for m in data.get("data", []) + ): + return data + last = ( + f"model not listed yet: " + f"{[m.get('id') for m in data.get('data', [])]}" + ) + time.sleep(2) + raise ValidationError(f"model readiness timed out: {last}") + + def discovery_instance_ids(self) -> set[str]: + """Instance IDs registered in the file discovery store (decimal strings). + + Dynamo's file store names model/instance records by hex instance ID, so + the directory itself lists every registered worker. + """ + if self.discovery is None: + return set() + found: set[str] = set() + root = Path(self.discovery.name) + for path in root.rglob("*"): + if not path.is_file(): + continue + key = unquote(path.relative_to(root).as_posix()) + parts = key.lower().split("/") + if "instances" not in parts and "models" not in parts: + continue + if len(parts) < 2 or parts[-2] != "generate": + continue + if re.fullmatch(r"[0-9a-f]{1,16}", parts[-1]): + found.add(str(int(parts[-1], 16))) + return found + + def discover_worker(self, client: Client, index: int) -> str: + """Learn the instance ID of worker[index]. + + Workers start one at a time, so the first unknown decode_worker_id + belongs to the worker most recently launched. Unhinted probes let the + router choose; when it keeps choosing a known worker, candidates from + the discovery store are confirmed with a pinned probe instead. + """ + known = {wid for i, wid in self.worker_ids.items() if i != index} + known |= self.retired_ids + errors = 0 + attempt = 0 + deadline = time.monotonic() + self.settings.readiness_timeout + while time.monotonic() < deadline: + attempt += 1 + self.deadline_check() + self.check_children() + candidates = [ + wid for wid in self.discovery_instance_ids() if wid not in known + ] + pinned = candidates[0] if attempt % 2 and len(candidates) == 1 else None + probe = client.completion( + f"probe-w{index}-{attempt:02d}", + f"identity probe {uuid.uuid4().hex}", + max_tokens=1, + extra_fields=["worker_id"], + worker=pinned, + ) + if probe.ok and probe.body is not None: + worker, _ = worker_identity(probe.body, "decode") + if worker is not None and worker not in known: + self.worker_ids[index] = worker + return worker + else: + errors += 1 + time.sleep(3) + raise ValidationError( + f"worker {index} was never selected by the router in {attempt} probes " + f"({errors} probe errors; discovery candidates " + f"{sorted(self.discovery_instance_ids() - known)})" + ) + + # -- teardown ---------------------------------------------------------- + + def wait_log( + self, child: Managed, needle: str, timeout: float, count: int = 1 + ) -> str: + deadline = time.monotonic() + timeout + while time.monotonic() < deadline: + self.deadline_check() + text = child.log_text() + if text.count(needle) >= count: + return text + if not child.alive(): + raise ValidationError( + f"{child.name} exited while waiting for {needle!r}" + ) + time.sleep(1) + raise ValidationError(f"{child.name} did not log {needle!r} within {timeout}s") + + def stop(self, *children: Managed) -> list[dict[str, Any]]: + outcomes = [] + targets = list(children) or list(reversed(self.children)) + for child in targets: + if not child.stopped: + outcomes.append(child.stop()) + return outcomes + + def cleanup(self) -> dict[str, Any]: + outcomes = self.stop() + self.reset_discovery() + removed = [] + for path in (self.kvcr_run_dir, self.pool_dir): + if path.exists(): + for item in sorted(path.rglob("*"), reverse=True): + try: + item.unlink() if not item.is_dir() else item.rmdir() + except OSError: + pass + try: + path.rmdir() + removed.append(str(path)) + except OSError: + pass + report = {"owned_processes": outcomes, "removed_paths": removed} + save_json(self.evidence / "cleanup.json", report) + if any("error" in outcome for outcome in outcomes): + raise ValidationError("owned process cleanup did not finish") + return report diff --git a/tests/e2e/phase1/telemetry.py b/tests/e2e/phase1/telemetry.py new file mode 100644 index 00000000..6646b461 --- /dev/null +++ b/tests/e2e/phase1/telemetry.py @@ -0,0 +1,270 @@ +# SPDX-License-Identifier: Apache-2.0 +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +"""Parse vLLM reset-interval KV transfer metrics from one worker's log. + +Each Phase 1 worker runs one data-parallel rank, so every metric line in a log +belongs to that worker. Values are interval samples; a window sums the samples +whose timestamps fall inside [start, end]. Nothing is subtracted between lines. +""" + +from __future__ import annotations + +import ast +import re +from dataclasses import dataclass +from datetime import datetime, timedelta, timezone +from pathlib import Path +from typing import Any + +from .common import ANSI + +# "ConnectorName={...}" segments of a MultiConnector metrics line; the dict +# body has no nested braces, so a greedy match up to the closing brace works. +CONNECTOR_DICT = re.compile(r"(\b[A-Za-z]+Connector)=(\{[^}]*\})") + +METRIC = re.compile( + r"(vllm:[^=]+)=([-+]?(?:\d+(?:\.\d*)?|\.\d+)(?:[eE][-+]?\d+)?)(?:, |$)" +) + + +def metric(name: str, *labels: str) -> str: + """Build the log form of a labelled vLLM metric, e.g. ('source_write',).""" + inner = ", ".join(repr(label) for label in labels) + if len(labels) == 1: + inner += "," + return f"vllm:{name}:({inner})" + + +def duration_count(scope: str, result: str = "success") -> str: + return metric("kvcr_duration_seconds", scope, result) + "_count" + + +TRANSFER_BLOCKS = "kvcr_transfer_blocks" +TRANSFER_BYTES = "kvcr_transfer_bytes" +TIER_READ_BYTES = metric("kv_offload_tiering_read_bytes", "1:kvcr") +TIER_WRITE_BYTES = metric("kv_offload_tiering_write_bytes", "1:kvcr") +STATE = "kvcr_state" + +SOURCE_WRITE = "source_write" +REMOTE_DELIVER = "remote_deliver" +LOCAL_FILL = "local_fill" +LOCAL_DELIVER = "local_deliver" +TRANSFER = "transfer" + + +VLLM_STAMP = re.compile(r"\b(\d\d)-(\d\d) (\d\d):(\d\d):(\d\d)\b") + + +def parse_timestamp(stamp: str) -> datetime | None: + try: + return datetime.fromisoformat(stamp.replace("Z", "+00:00")) + except ValueError: + return None + + +ISO_INSTANT = re.compile( + r"^\d{4}-\d\d-\d\dT\d\d:\d\d:\d\d(?:\.\d+)?(?:Z|[+-]\d\d:\d\d)$" +) + + +def precise_stamp(line: str) -> datetime | None: + """Timezone-aware ISO instant at the start of a Dynamo-formatted line. + + Only a full instant with a zone counts: a bare date or a naive time at the + start of some other line must not become a comparison partner for the + aware request timestamps. + """ + head = line.split(maxsplit=1)[0] if line else "" + if not ISO_INSTANT.match(head): + return None + stamp = parse_timestamp(head) + return stamp if stamp is not None and stamp.tzinfo is not None else None + + +def line_timestamp(line: str) -> datetime | None: + """Timestamp of a metrics line in either log format. + + Dynamo's tracing formatter starts the line with an ISO instant + (``2026-10-03T21:41:22.673433Z ...``); vLLM's own logger, which Dynamo + ea23e51+ lets write the KV Transfer metrics line, prefixes + ``(MainProcess pid=N) INFO 10-08 20:54:47 [...]`` with no year or zone. + The latter is taken as UTC of the current year (the harness runs with + TZ=UTC containers); a run spanning New Year would need the launch date. + """ + stamp = parse_timestamp(line.split()[0]) + if stamp is not None: + return stamp + head = line.split("KV Transfer metrics:", 1)[0] + found = VLLM_STAMP.search(head) + if found is None: + return None + month, day, hour, minute, second = (int(x) for x in found.groups()) + now = datetime.now(timezone.utc) + year = now.year + candidate = datetime(year, month, day, hour, minute, second, tzinfo=timezone.utc) + if candidate - now > timedelta(days=1): # log from late last year + candidate = candidate.replace(year=year - 1) + return candidate + + +@dataclass(frozen=True) +class Snapshot: + timestamp: datetime + line: int + metrics: dict[str, int | float] + # vLLM's own logger stamps to the second; Dynamo's formatter to the + # microsecond. For a second-precision sample, preceded_by is the latest + # microsecond-precision stamp seen earlier in the log (a request's + # handle_payload line, typically): it tells whether the sample was flushed + # after a request that started in the same second. + second_precision: bool = False + preceded_by: datetime | None = None + + +def parse_log(text: str) -> tuple[list[Snapshot], list[str]]: + """Return interval snapshots and parse errors for complete log lines.""" + snapshots: list[Snapshot] = [] + errors: list[str] = [] + last_precise: datetime | None = None + for number, raw in enumerate(text.split("\n")[:-1], 1): + line = ANSI.sub("", raw).strip() + precise = precise_stamp(line) + if precise is not None and (last_precise is None or precise > last_precise): + last_precise = precise + if "KV Transfer metrics:" not in line: + continue + stamp = line_timestamp(line) + if stamp is None: + errors.append(f"line {number}: metric line without a parseable timestamp") + continue + payload = line.split("KV Transfer metrics:", 1)[1].strip() + values: dict[str, int | float] = {} + if CONNECTOR_DICT.search(payload): + # MultiConnector: one Python dict repr per connector, + # "OffloadingConnector={...}, NixlConnector={...}". Merge them; the + # KVCR tier lives in the OffloadingConnector dict. + parsed = 0 + for found in CONNECTOR_DICT.finditer(payload): + try: + block = ast.literal_eval(found.group(2)) + except (ValueError, SyntaxError): + continue + if isinstance(block, dict): + parsed += 1 + for name, value in block.items(): + if isinstance(value, (int, float)) and not isinstance( + value, bool + ): + values[str(name)] = value + if not parsed: + errors.append(f"line {number}: unparsable per-connector metrics") + continue + else: + pairs = METRIC.findall(payload) + for name, value in pairs: + values[name] = ( + float(value) if any(c in value for c in ".eE") else int(value) + ) + if not values or len(values) != payload.count("="): + errors.append(f"line {number}: incomplete metric parsing") + continue + snapshots.append(Snapshot(stamp, number, values, precise is None, last_precise)) + return snapshots, errors + + +@dataclass +class Window: + """Summed interval samples between two instants for one worker.""" + + totals: dict[str, int | float] + samples: int + first_after_end: bool + peer_failures: list[dict[str, Any]] + parse_errors: list[str] + + def get(self, name: str, default: int | float = 0) -> int | float: + return self.totals.get(name, default) + + def blocks(self, scope: str) -> int | float: + return self.get(metric(TRANSFER_BLOCKS, scope)) + + def bytes(self, scope: str) -> int | float: + return self.get(metric(TRANSFER_BYTES, scope)) + + def successes(self, scope: str) -> int | float: + return self.get(duration_count(scope)) + + def gauge(self, resource: str) -> int | float | None: + value = self.totals.get(metric(STATE, resource)) + return value + + +def flushed_after_start(snapshot: Snapshot, start: datetime) -> bool: + """Whether a second-precision sample stamped in start's second was flushed + after start. + + vLLM's logger stamps metric lines to the second, so a sample flushed later + in the second a request started in parses as earlier than the request. It + counts for the request if a microsecond-stamped line (Dynamo's request + handling) stamped at or after start precedes it in the log; a sample flushed + before the request has no such line ahead of it and belongs to the previous + request. + """ + return ( + snapshot.second_precision + and snapshot.timestamp == start.replace(microsecond=0) + and snapshot.preceded_by is not None + and snapshot.preceded_by >= start + ) + + +def window( + snapshots: list[Snapshot], + errors: list[str], + start: datetime, + end: datetime | None = None, +) -> Window: + """Sum samples stamped at or after start (and at or before end). + + first_after_end reports whether a sample past `end` exists, which shows the + interval containing the request has been flushed by the engine. + """ + totals: dict[str, int | float] = {} + failures = [] + count = 0 + flushed = False + for snapshot in snapshots: + if snapshot.timestamp < start and not flushed_after_start(snapshot, start): + continue + if end is not None and snapshot.timestamp > end: + flushed = True + continue + count += 1 + for name, value in snapshot.metrics.items(): + if name.startswith(f"vllm:{STATE}:"): + totals[name] = value # gauge: keep the latest sample + else: + totals[name] = totals.get(name, 0) + value + if ( + value > 0 + and name.startswith("vllm:kvcr_") + and ("'failed'" in name or "'partial'" in name) + ): + failures.append({"metric": name, "value": value, "line": snapshot.line}) + if errors: + totals["_parse_errors"] = len(errors) + return Window(totals, count, flushed, failures, list(errors)) + + +def read_window(log: Path, start: datetime, end: datetime | None = None) -> Window: + snapshots, errors = parse_log(log.read_text(errors="replace")) + return window(snapshots, errors, start, end) + + +def latest_gauge(log: Path, resource: str) -> int | float | None: + snapshots, _ = parse_log(log.read_text(errors="replace")) + name = metric(STATE, resource) + for snapshot in reversed(snapshots): + if name in snapshot.metrics: + return snapshot.metrics[name] + return None diff --git a/tests/e2e/run_phase1.py b/tests/e2e/run_phase1.py new file mode 100644 index 00000000..f0bf966f --- /dev/null +++ b/tests/e2e/run_phase1.py @@ -0,0 +1,584 @@ +# SPDX-License-Identifier: Apache-2.0 +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +"""Run KVCR test plan v7 Phase 1 on one node with two GPUs. + +Sequence: + 1. preflight GPUs, shared memory, kernel (Guard), ports + 2. reference KVCR disabled worker; every prompt generated twice + 3. stack per-worker kvcr_service (Guard) + KV router + 2 workers + 4. cases the selected TC-* cases (TC-01 in this revision) + 5. evidence result.json, junit.xml, all requests/responses/logs + +Exit status is 0 only when every selected case is PASS. BLOCKED cases are +reported as skipped with the missing condition and make the run fail unless +--allow-blocked is given. +""" + +from __future__ import annotations + +import argparse +import hashlib +import importlib.metadata +import json +import os +import platform +import re +import signal +import socket +import sys +import time +import uuid +from pathlib import Path +from typing import Any + +sys.path.insert(0, str(Path(__file__).resolve().parent)) + +from phase1.cases import CASES, Context, Prompts, run_case # noqa: E402 +from phase1.client import Client # noqa: E402 +from phase1.common import ( # noqa: E402 + BLOCKED, + FAIL, + PASS, + BlockedError, + TokenSample, + ValidationError, + compare_tokens, + extract_sample, + save_json, + utc_now, + write_junit, +) +from phase1.stack import GUARD, MODES, Settings, Stack # noqa: E402 + + +def provenance(args: argparse.Namespace) -> dict[str, Any]: + versions = {} + for distribution in ( + "kvcr", + "nixl", + "vllm", + "ai-dynamo", + "ai-dynamo-runtime", + "torch", + ): + try: + versions[distribution] = importlib.metadata.version(distribution) + except importlib.metadata.PackageNotFoundError: + versions[distribution] = None + manifest = Path("/opt/kvcr-build/source-revisions.json") + package = Path(__file__).resolve().parent / "phase1" + digest = hashlib.sha256() + for path in sorted(package.glob("*.py")) + [Path(__file__).resolve()]: + digest.update(path.name.encode()) + digest.update(path.read_bytes()) + return { + "python": sys.version, + "platform": platform.platform(), + "versions": versions, + "arguments": { + key: str(value) if isinstance(value, Path) else value + for key, value in vars(args).items() + }, + "image_source_revisions": json.loads(manifest.read_text()) + if manifest.exists() + else None, + "harness_sha256": digest.hexdigest(), + } + + +class Runner: + def __init__(self, args: argparse.Namespace) -> None: + self.args = args + self.output = args.output_dir.resolve() + if self.output.exists(): + stray = [p for p in self.output.rglob("*") if p.is_file()] + if stray: + raise ValidationError( + f"output directory must be empty: {self.output} (found {stray[0]})" + ) + self.output.mkdir(parents=True, exist_ok=True) + self.deadline = time.monotonic() + args.timeout + self.settings = Settings( + model=args.model, + output=self.output, + mode=args.mode, + gpus=tuple(args.gpus), + http_port=args.http_port, + reference_http_port=args.reference_http_port, + pool_gb=args.pool_gb, + compatibility_digest=args.compatibility_digest or "", + gpu_memory_utilization=args.gpu_memory_utilization, + max_model_len=args.max_model_len, + cpu_bytes_to_use=args.cpu_bytes_to_use, + batch_invariant=args.batch_invariant, + state_agent=not args.no_state_agent, + readiness_timeout=args.readiness_timeout, + request_timeout=args.request_timeout, + extra_worker_args=list(args.worker_arg), + frontend_env=dict( + item.split("=", 1) for item in args.frontend_env if "=" in item + ), + disable_hybrid_kv_cache_manager=not args.keep_hybrid_kv_cache_manager, + enforce_eager=not args.no_enforce_eager, + ) + self.stack = Stack(self.settings, self.output / "stack") + self.stack.deadline_check = self.check_deadline + self.tag = args.prompt_tag or uuid.uuid4().hex[:12] + self.prompts = Prompts(self.tag) + self.references: dict[str, TokenSample] = {} + # cached_tokens the KVCR-disabled engine reported per reference request, + # in order; tells a case what the engine alone can reuse. + self.reference_cached: dict[str, list[int | None]] = {} + self.result: dict[str, Any] = { + "plan": "KVCR Test Plan v7, Phase 1 (Guard-mode critical path)", + "started_at": utc_now(), + "status": "running", + "mode": args.mode, + "deployment": "aggregated", + "model": args.model, + "prompt_tag": self.tag, + "max_tokens": args.max_tokens, + "hit_block_tokens": args.hit_block_tokens, + "worker_args": list(args.worker_arg), + "disable_hybrid_kv_cache_manager": not args.keep_hybrid_kv_cache_manager, + "enforce_eager": not args.no_enforce_eager, + "batch_invariant": args.batch_invariant, + "state_agent": not args.no_state_agent, + "selected_cases": list(args.cases), + "cases": [], + "reference": {}, + "limitations": [ + "Single node, two GPUs, aggregated deployment only.", + "Local-hit path evidence is the KVCR commit (local_fill) after the " + "cold request plus the reuse hit count.", + ], + } + self.save() + + def save(self) -> None: + save_json(self.output / "result.json", self.result) + + def check_deadline(self) -> None: + if time.monotonic() > self.deadline: + raise ValidationError("overall run deadline expired") + + # -- phases ------------------------------------------------------------ + + def reference_phase(self) -> None: + """Establish reference tokens from a KVCR-disabled engine, twice each.""" + evidence = self.output / "reference" + evidence.mkdir() + stack = self.stack + stack.start_frontend( + self.settings.reference_http_port, name="reference-frontend" + ) + worker = stack.start_worker(0, kvcr=False, tag="-reference") + client = Client( + f"http://127.0.0.1:{self.settings.reference_http_port}", + self.args.model, + evidence, + max_tokens=self.args.max_tokens, + request_timeout=self.args.request_timeout, + ) + stack.wait_model(client) + stack.discover_worker(client, 0) + launch = json.loads((stack.evidence / f"{worker.name}.launch.json").read_text()) + bypass = not any( + "OffloadingConnector" in part or '"kvcr"' in part + for part in launch["command"] + ) + samples: dict[str, list[TokenSample]] = {} + unstable = [] + unstable_detail: dict[str, Any] = {} + # Only the prompts the requested cases use: a long prompt of a case not + # being run must not block the run when the bare engine is unstable on it. + case_prompts = {"TC-01": {"tc01"}} + wanted: set[str] = set() + for case in self.args.cases: + wanted |= case_prompts.get(case, set()) + selected = { + name: prompt + for name, prompt in self.prompts.all().items() + if name in wanted + } + for name, prompt in selected.items(): + runs = [] + for repeat in (1, 2): + response = client.completion( + f"ref-{name}-{repeat}", prompt, **stack.pin(0) + ) + if not response.ok or response.body is None: + raise ValidationError( + f"reference request {name} failed: {response.error}" + ) + runs.append(extract_sample(response.body)) + details = (response.body.get("usage") or {}).get( + "prompt_tokens_details" + ) or {} + cached_value = details.get("cached_tokens") + self.reference_cached.setdefault(name, []).append( + cached_value if type(cached_value) is int else None + ) + samples[name] = runs + # The non-streaming runs carry prompt IDs; streamed runs may not. + canonical = runs[-1] + labels = ["1", "2"] + divergence = {} + for label, run in zip(labels, runs): + problems = [ + p + for p in compare_tokens(canonical, run) + if "prompt token IDs" not in p + ] + if problems: + first = next( + ( + i + for i, (x, y) in enumerate( + zip( + canonical.completion_token_ids, + run.completion_token_ids, + ) + ) + if x != y + ), + min(canonical.length, run.length), + ) + divergence[label] = { + "first_divergent_token": first, + "problems": problems, + } + if divergence: + unstable.append(name) + unstable_detail[name] = divergence + else: + self.references[name] = canonical + # Filesystem paths the harness itself chose may contain "kvcr" (an HF + # cache under /cache/doxu-kvcr/, the evidence tree); they say nothing + # about the engine, so blank them before counting mentions. + own_names = [ + str(self.output), + os.environ.get("HF_HOME", ""), + socket.gethostname(), + ] + scrubbed = worker.log_text() + for name in sorted( + {n for n in own_names if n and "kvcr" in n.lower()}, key=len, reverse=True + ): + scrubbed = scrubbed.replace(name, "") + kvcr_mentions = len(re.findall(r"(?i)kvcr", scrubbed)) + kvcr_lines = [ + line[:200] for line in scrubbed.splitlines() if re.search(r"(?i)kvcr", line) + ][:10] + self.result["reference"] = { + "kvcr_disabled_in_launch": bypass, + "batch_invariant": self.settings.batch_invariant, + "unstable": unstable_detail, + "kvcr_mentions_in_worker_log": kvcr_mentions, + "cached_tokens": self.reference_cached, + "kvcr_lines_in_worker_log": kvcr_lines, + "prompts": { + name: { + "prompt_tokens": max(len(run.prompt_token_ids) for run in runs), + "completion_tokens": [run.length for run in runs], + "finish_reason": [run.finish_reason for run in runs], + "stable": name not in unstable, + } + for name, runs in samples.items() + }, + } + save_json( + evidence / "reference-tokens.json", + {name: sample.__dict__ for name, sample in self.references.items()}, + ) + self.save() + stack.stop(*[c for c in (worker, stack.frontend) if c is not None]) # type: ignore[arg-type] + # Forget the reference handles so the cases phase does not see the + # stopped reference worker. + stack.workers.pop(0, None) + stack.worker_ids.pop(0, None) + stack.reset_discovery() + if not bypass or kvcr_mentions: + raise ValidationError("reference engine was not verified KVCR-free") + if unstable: + raise ValidationError(f"reference generation unstable for: {unstable}") + + def stack_phase(self) -> Client: + stack = self.stack + client = Client( + f"http://127.0.0.1:{self.settings.http_port}", + self.args.model, + self.output / "requests", + max_tokens=self.args.max_tokens, + request_timeout=self.args.request_timeout, + ) + client.evidence.mkdir() + if self.settings.mode == GUARD: + for index in (0, 1): + stack.start_service(index) + if stack.uses_state_agent(): + stack.start_state_agent() + stack.start_frontend(self.settings.http_port) + for index in (0, 1): + stack.start_worker(index) + stack.wait_model(client) + stack.discover_worker(client, index) + self.result["workers"] = { + str(index): {"instance_id": wid, "gpu": self.settings.gpus[index]} + for index, wid in stack.worker_ids.items() + } + self.result["compatibility_digest"] = self.settings.compatibility_digest + self.save() + return client + + def cases_phase(self, client: Client) -> None: + ctx = Context( + self.stack, + client, + self.prompts, + self.references, + index_wait=self.args.index_wait, + telemetry_timeout=self.args.telemetry_timeout, + hit_block_tokens=self.args.hit_block_tokens, + ) + for name in self.args.cases: + self.check_deadline() + result = run_case(name, ctx) + self.result["cases"].append(result.to_dict()) + self.result["requests"] = client.log + self.save() + print( + json.dumps( + {"case": name, "status": result.status, "error": result.error} + ) + ) + if result.status == FAIL and self.args.stop_on_fail: + break + + # -- lifecycle --------------------------------------------------------- + + def run(self) -> int: + blocked_all: str | None = None + try: + save_json(self.output / "provenance.json", provenance(self.args)) + self.result["preflight"] = self.stack.preflight() + self.reference_phase() + client = self.stack_phase() + self.cases_phase(client) + except BlockedError as blocked: + blocked_all = str(blocked) + except (Exception, KeyboardInterrupt) as error: + self.result["error"] = f"{type(error).__name__}: {error}" + finally: + try: + self.stack.cleanup() + except Exception as error: # noqa: BLE001 + self.result["cleanup_error"] = str(error) + if blocked_all is not None: + for name in self.args.cases: + if not any(case["case"] == name for case in self.result["cases"]): + self.result["cases"].append( + { + "case": name, + "title": name, + "requirement": "", + "status": BLOCKED, + "checks": [], + "evidence": [], + "error": blocked_all, + "started_at": utc_now(), + "finished_at": utc_now(), + "measurements": {}, + } + ) + statuses = {case["status"] for case in self.result["cases"]} + missing = [ + n + for n in self.args.cases + if not any(c["case"] == n for c in self.result["cases"]) + ] + if ( + missing + or "error" in self.result + or "cleanup_error" in self.result + or FAIL in statuses + ): + self.result["status"] = FAIL + elif BLOCKED in statuses: + self.result["status"] = BLOCKED + else: + self.result["status"] = PASS + self.result["missing_cases"] = missing + self.result["finished_at"] = utc_now() + self.save() + write_junit(self.output, self.result) + print( + json.dumps( + { + "status": self.result["status"], + "cases": {c["case"]: c["status"] for c in self.result["cases"]}, + "error": self.result.get("error"), + "result": str(self.output / "result.json"), + } + ) + ) + if self.result["status"] == PASS: + return 0 + if self.result["status"] == BLOCKED and self.args.allow_blocked: + return 0 + return 1 + + +def arguments(argv: list[str] | None = None) -> argparse.Namespace: + parser = argparse.ArgumentParser( + description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter + ) + parser.add_argument("--output-dir", required=True, type=Path) + parser.add_argument("--model", default="Qwen/Qwen3-0.6B") + parser.add_argument("--mode", choices=MODES, default=GUARD) + parser.add_argument( + "--gpus", + default="0,1", + help="two CUDA device indices, e.g. 0,1", + ) + parser.add_argument("--cases", default="TC-01") + parser.add_argument("--http-port", type=int, default=8000) + parser.add_argument("--reference-http-port", type=int, default=8001) + parser.add_argument( + "--pool-gb", type=int, default=2, help="Guard pool size per worker" + ) + parser.add_argument("--compatibility-digest", default=None) + parser.add_argument( + "--gpu-memory-utilization", + type=float, + default=0.8, + help="vLLM --gpu-memory-utilization per worker (<= 0.45 when both share " + "one GPU)", + ) + parser.add_argument("--max-tokens", type=int, default=32) + parser.add_argument( + "--hit-block-tokens", + type=int, + default=64, + help="tokens per prefix-cache hit block: 64 for dense models, 576 for Qwen3.5 " + "hybrid models (vLLM raises the attention block); hit alignment and the " + "delivered-pages checks use it", + ) + parser.add_argument( + "--frontend-env", + action="append", + default=[], + metavar="KEY=VALUE", + help="environment for the Dynamo frontend process only, e.g. " + "DYN_LOG=info,dynamo_kv_router=debug,dynamo_llm::kv_router=debug", + ) + parser.add_argument( + "--cpu-bytes-to-use", + type=int, + default=2_000_000_000, + help="vLLM host primary tier bytes per worker (OffloadingConnector " + "cpu_bytes_to_use)", + ) + parser.add_argument( + "--max-model-len", + type=int, + default=32000, + help="vLLM --max-model-len for every engine (lower it together with " + "--worker-arg=--num-gpu-blocks-override=N, vLLM requires N*block_size >= this)", + ) + parser.add_argument( + "--keep-hybrid-kv-cache-manager", + action="store_true", + help="do not pass --disable-hybrid-kv-cache-manager to vLLM (needed for " + "SSM/linear-attention hybrids such as Qwen3.5)", + ) + parser.add_argument( + "--no-enforce-eager", + action="store_true", + help="let vLLM use CUDA graphs (much faster decode for GDN/Mamba hybrids)", + ) + parser.add_argument( + "--worker-arg", + action="append", + default=[], + help="extra dynamo.vllm argument, repeatable (e.g. " + "--worker-arg=--mamba-cache-mode=align)", + ) + parser.add_argument( + "--prompt-tag", default=None, help="Fixed tag for reproducible prompts" + ) + parser.add_argument( + "--batch-invariant", action="store_true", help="Set VLLM_BATCH_INVARIANT=1" + ) + parser.add_argument( + "--no-state-agent", + action="store_true", + help="Guard mode: do not run dynamo.kv_state_agent (no router hints)", + ) + parser.add_argument( + "--timeout", type=float, default=2400, help="Overall deadline in seconds" + ) + parser.add_argument("--readiness-timeout", type=float, default=600) + parser.add_argument("--request-timeout", type=float, default=120) + parser.add_argument("--telemetry-timeout", type=float, default=60) + parser.add_argument("--index-wait", type=float, default=15) + parser.add_argument("--stop-on-fail", action="store_true") + parser.add_argument( + "--allow-blocked", action="store_true", help="Exit 0 when only BLOCKED remains" + ) + args = parser.parse_args(argv) + gpus = [part.strip() for part in args.gpus.split(",")] + if len(gpus) != 2 or not all(part.isdecimal() for part in gpus): + parser.error("--gpus needs two CUDA device indices, e.g. 0,1") + args.gpus = [int(part) for part in gpus] + if args.gpus[0] == args.gpus[1] and args.gpu_memory_utilization > 0.45: + parser.error("two workers on one GPU need --gpu-memory-utilization <= 0.45") + if not 0.05 <= args.gpu_memory_utilization <= 0.95: + parser.error("--gpu-memory-utilization must be within 0.05..0.95") + cases = [part.strip().upper() for part in args.cases.split(",") if part.strip()] + unknown = [case for case in cases if case not in CASES] + if unknown or not cases: + parser.error(f"unknown cases {unknown}; choose from {sorted(CASES)}") + args.cases = cases + for name in ( + "timeout", + "readiness_timeout", + "request_timeout", + "telemetry_timeout", + ): + if not 0 < getattr(args, name) <= 7200: + parser.error(f"--{name.replace('_', '-')} must be >0 and <=7200 seconds") + if not 0 <= args.index_wait <= 300: + parser.error("--index-wait must be between 0 and 300 seconds") + if not 1 <= args.max_tokens <= 512 or not 1 <= args.pool_gb <= 64: + parser.error("--max-tokens must be 1..512 and --pool-gb 1..64") + if args.hit_block_tokens < 1 or args.hit_block_tokens % 64: + parser.error("--hit-block-tokens must be a positive multiple of 64") + if args.http_port == args.reference_http_port: + parser.error("--http-port and --reference-http-port must differ") + return args + + +def main() -> int: + args = arguments() + try: + runner = Runner(args) + except Exception as error: # noqa: BLE001 + print(f"Cannot initialize run: {error}", file=sys.stderr) + return 1 + + def interrupted(signum: int, _frame: Any) -> None: + raise KeyboardInterrupt(f"received signal {signum}") + + previous = { + n: signal.signal(n, interrupted) for n in (signal.SIGINT, signal.SIGTERM) + } + try: + return runner.run() + finally: + for number, handler in previous.items(): + signal.signal(number, handler) + + +if __name__ == "__main__": + raise SystemExit(main())