[integration_test] add mutation meta-test and four vllm e2e scenarios

* test_mutation (B4): runs the basic scenario with MutatedConnector (slot -1
  in _attn_token_indices) and asserts KV verification FAILS -- proof the
  capture-based harness catches symmetric translation bugs and is not vacuous.
* test_full_hit (C2): prompt trimmed to an exact multiple of the manager
  block size, resent after being fully saved. Regression for the synchronous
  full-hit crash (vllm 0.26.0 scheduler.py 'assert num_new_tokens > 0');
  asserts the engine survives and 0 < matched < prompt tokens.
* test_partial_hit (C1): staged A / A+B / A+B+C prompts with prefix caching
  on; exercises the non-zero-offset incremental manager query and the
  incremental save extension, asserts a logged query with offset > 0 and
  verifies the blocks saved through the incremental path.
* test_load_failure (C3): deletes the tail half of the per-block storage
  files between save and load (key_count_per_file=1, block_per_load_task=1,
  kv_load_failure_policy=recompute). Full-attn: failures reported to vLLM,
  surviving head blocks verify bit-exact, mismatches confined to deleted
  blocks. Hybrid: failure swallowed by design (vLLM invalid-block recovery is
  single-group only), asserts no hang/crash and the failure log.
* test_multi_turn (C4): turn 1 decodes past a manager block boundary
  (ignore_eos + return_token_ids), asserts the manager committed more blocks
  than the prompt covers; turn 2 embeds turn 1 prompt+output as token ids and
  must externally match beyond prompt-only coverage with verified KV.

All scenarios pass for both Qwen2.5-7B-Instruct (full-attn) and Qwen3.5-4B
(hybrid) alongside the original basic/concurrent/tp regressions.

Author: lpdink

integration_test/vllm_e2e/BUILD

@@ -59,3 +59,85 @@ py_test(
     timeout = "eternal",
     deps = [":e2e_lib"],
 )
+
+py_test(
+    name = "test_full_hit",
+    srcs = ["test_full_hit.py"],
+    data = [
+        "//kv_cache_manager:kv_cache_manager_bin",
+    ],
+    imports = ["."],
+    tags = [
+        "no-remote-exec",
+        "gpu",  # requires 1+ GPU
+        "exclusive",  # GPU tests must run serially to avoid CUDA OOM contention
+    ],
+    timeout = "eternal",
+    deps = [":e2e_lib"],
+)
+
+py_test(
+    name = "test_partial_hit",
+    srcs = ["test_partial_hit.py"],
+    data = [
+        "//kv_cache_manager:kv_cache_manager_bin",
+    ],
+    imports = ["."],
+    tags = [
+        "no-remote-exec",
+        "gpu",  # requires 1+ GPU
+        "exclusive",  # GPU tests must run serially to avoid CUDA OOM contention
+    ],
+    timeout = "eternal",
+    deps = [":e2e_lib"],
+)
+
+py_test(
+    name = "test_load_failure",
+    srcs = ["test_load_failure.py"],
+    data = [
+        "//kv_cache_manager:kv_cache_manager_bin",
+    ],
+    imports = ["."],
+    tags = [
+        "no-remote-exec",
+        "gpu",  # requires 1+ GPU
+        "exclusive",  # GPU tests must run serially to avoid CUDA OOM contention
+    ],
+    timeout = "eternal",
+    deps = [":e2e_lib"],
+)
+
+py_test(
+    name = "test_multi_turn",
+    srcs = ["test_multi_turn.py"],
+    data = [
+        "//kv_cache_manager:kv_cache_manager_bin",
+    ],
+    imports = ["."],
+    tags = [
+        "no-remote-exec",
+        "gpu",  # requires 1+ GPU
+        "exclusive",  # GPU tests must run serially to avoid CUDA OOM contention
+    ],
+    timeout = "eternal",
+    deps = [":e2e_lib"],
+)
+
+# Meta-test: injects an off-by-one into the connector's token translation and
+# asserts the KV verification FAILS -- proof the harness is not vacuous.
+py_test(
+    name = "test_mutation",
+    srcs = ["test_mutation.py"],
+    data = [
+        "//kv_cache_manager:kv_cache_manager_bin",
+    ],
+    imports = ["."],
+    tags = [
+        "no-remote-exec",
+        "gpu",  # requires 1+ GPU
+        "exclusive",  # GPU tests must run serially to avoid CUDA OOM contention
+    ],
+    timeout = "eternal",
+    deps = [":e2e_lib"],
+)

integration_test/vllm_e2e/test_full_hit.py

@@ -0,0 +1,82 @@
+"""test_full_hit: full-prompt external hit must not crash the engine.
+
+Regression test for the synchronous-load full-hit bug: this connector reports
+external matches with load_kv_async=False, so vLLM schedules
+``num_tokens - num_computed_tokens`` new tokens and asserts that count is > 0
+(vllm/v1/core/sched/scheduler.py, waiting-queue loop: ``assert num_new_tokens
+> 0``). Without capping, a prompt whose token count is an exact multiple of the
+manager block size and whose blocks are all externally cached would make the
+count 0 and kill the engine.
+
+Phase 1 saves a prompt of exactly N manager blocks; phase 2 resends the very
+same prompt (as explicit token ids, so tokenization cannot shift the length).
+Asserts: the engine survives, the completion is well-formed, and the connector
+reports 0 < matched < prompt tokens (the cap dropped at least the last block).
+
+Runs against both full-attention and hybrid models via $KVCM_E2E_MODEL.
+"""
+
+import logging
+import unittest
+
+from e2e_lib import (
+    ScenarioEnv, make_base_prompts, send_completions, tokenize,
+    wait_for_prefix_cached,
+)
+
+logger = logging.getLogger("vllm_e2e")
+
+
+class TestFullHit(unittest.TestCase):
+    def test_full_hit(self):
+        env = ScenarioEnv("full_hit")
+        try:
+            env.start_manager()
+            vllm = env.start_vllm(log_suffix="_p1" if env.hybrid else "")
+
+            mbs = env.manager_block_size()
+            # Trim a long-enough prompt's token ids to an exact multiple of the
+            # manager block size (>= 2 blocks so the cap has room to drop one).
+            toks = tokenize(vllm.base_url(), make_base_prompts(1, env.hybrid)[0])
+            num_blocks = len(toks) // mbs
+            self.assertGreaterEqual(
+                num_blocks, 2, f"prompt too short: {len(toks)} tokens, mbs={mbs}")
+            prompt_ids = toks[:num_blocks * mbs]
+            logger.info("full-hit prompt: %d tokens = %d x %d",
+                        len(prompt_ids), num_blocks, mbs)
+
+            # Phase 1: fresh prefill -> all blocks saved.
+            resp1 = send_completions(vllm.base_url(), [prompt_ids])[0]
+            self.assertTrue(resp1["choices"][0]["text"])
+            self.assertTrue(wait_for_prefix_cached(
+                env.manager.manager_uri(), env.instance_id, prompt_ids,
+                min_blocks=num_blocks))
+
+            if env.hybrid:
+                # Clear the local prefix cache so phase 2 goes external.
+                vllm = env.restart_vllm(log_suffix="_p2")
+
+            # Phase 2: the exact same prompt -> full external hit. Without the
+            # cap this crashes the engine (assert num_new_tokens > 0).
+            resp2 = send_completions(vllm.base_url(), [prompt_ids])[0]
+            self.assertTrue(resp2["choices"][0]["text"])
+
+            # The engine must still be alive and serving.
+            resp3 = send_completions(vllm.base_url(), ["sanity check prompt"])[0]
+            self.assertTrue(resp3["choices"][0]["text"])
+
+            # Connector-side evidence: matched > 0 (external hit happened) and
+            # matched < prompt tokens (the cap left tokens to recompute).
+            matched = [int(g[0]) for g in
+                       env.scan_connector_logs(r"matched (\d+) external tokens")]
+            self.assertTrue(matched, "no 'matched N external tokens' log found")
+            hit = [m for m in matched if m > 0]
+            self.assertTrue(hit, f"no positive external match in {matched}")
+            self.assertTrue(all(m < len(prompt_ids) for m in hit),
+                            f"match not capped below prompt len: {matched}")
+        finally:
+            env.stop()
+
+
+if __name__ == "__main__":
+    unittest.main()

integration_test/vllm_e2e/test_load_failure.py

@@ -0,0 +1,161 @@
+"""test_load_failure: storage loss between save and load must degrade, not kill.
+
+Phase 1 saves a long prompt; the test then deletes the storage files of the
+*tail half* of the manager blocks (resolved through the manager's ordered
+getCacheLocation response, one file per block via key_count_per_file=1).
+Phase 2 reloads the same prefix with ``block_per_load_task=1`` so every block
+fails or succeeds independently.
+
+Full-attention models (single group, report_failures=True): the connector
+reports the failed blocks' vLLM block ids; with
+``kv_load_failure_policy="recompute"`` (vLLM 0.26.0 defaults to "fail", which
+turns any load failure into a 500) vLLM truncates the computed-token count at
+the first invalid block and recomputes from there
+(vllm/v1/core/sched/scheduler.py::_handle_invalid_blocks /
+_update_requests_with_invalid_blocks). Asserts: the request returns a normal
+completion, the failures were logged and reported, every *surviving* head
+block's loaded KV is bit-exact, and any mismatching capture belongs to a
+deleted (recomputed) block. Recomputed blocks are not held to bit-exactness:
+they contain freshly recomputed KV whose numerics depend on prefill chunking,
+which is vLLM's business, not the connector's.
+
+Hybrid models (multiple groups, report_failures=False): vLLM's invalid-block
+recovery only supports single-group block tables, so the connector only logs
+the failure. Asserts: the request still returns (no hang, no crash) and the
+failure was logged. KV content is NOT verified: with the failure swallowed
+the affected blocks keep garbage by design.
+
+This scenario also regression-tests the fail-reschedule loop fix: a request
+whose external load failed must not re-match external blocks on requeue
+(v1_connector.get_num_new_matched_tokens retry guard), otherwise the engine
+loops load-fail-reschedule forever and the request hangs.
+"""
+
+import logging
+import os
+import unittest
+from urllib.parse import urlparse
+
+import requests
+
+from e2e_lib import (
+    ScenarioEnv, compare_captures, full_block_hashes, make_base_prompts,
+    send_completions, tokenize, wait_for_captures, wait_for_prefix_cached,
+)
+
+logger = logging.getLogger("vllm_e2e")
+
+
+def get_block_files(manager_uri: str, instance_id: str, token_ids: list[int],
+                    spec_name: str = "tp0_g0") -> list[str]:
+    """Per-manager-block storage file paths, in block order, from the manager's
+    getCacheLocation response."""
+    r = requests.post(f"{manager_uri}/api/getCacheLocation", json={
+        "trace_id": "e2e_block_files",
+        "token_ids": token_ids,
+        "instance_id": instance_id,
+        "query_type": "QT_PREFIX_MATCH",
+        "block_mask": {"offset": 0},
+    }, timeout=10)
+    r.raise_for_status()
+    files = []
+    for location in r.json().get("locations", []):
+        for spec in location.get("location_specs", []):
+            if spec["name"] == spec_name:
+                # uri: file://<path>?size=...
+                files.append(urlparse(spec["uri"]).path)
+    return files
+
+
+class TestLoadFailure(unittest.TestCase):
+    def test_load_failure(self):
+        env = ScenarioEnv(
+            "load_failure",
+            extra_config_overrides={"block_per_load_task": 1},
+            key_count_per_file=1,  # one file per block -> per-block failures
+            kv_load_failure_policy="recompute",
+        )
+        try:
+            env.start_manager()
+            vllm = env.start_vllm(log_suffix="_p1" if env.hybrid else "")
+            mbs = env.manager_block_size()
+
+            prompt = make_base_prompts(1, env.hybrid)[0]
+            suffix = " Now answer: what is 2+2?"
+            toks = tokenize(vllm.base_url(), prompt)
+            save_blocks = len(toks) // mbs
+            self.assertGreaterEqual(save_blocks, 2)
+
+            # ---- Phase 1: save everything.
+            send_completions(vllm.base_url(), [prompt])
+            wait_for_captures(env.capture_dir, "ref", expected=save_blocks,
+                              timeout=180)
+            self.assertTrue(wait_for_prefix_cached(
+                env.manager.manager_uri(), env.instance_id, toks,
+                min_blocks=save_blocks))
+
+            # ---- Sabotage: delete the tail half of the blocks' files. The
+            # head blocks stay loadable, so vLLM truncates at the first deleted
+            # block and the surviving loads remain verifiable.
+            files = get_block_files(env.manager.manager_uri(), env.instance_id,
+                                    toks)
+            self.assertEqual(len(files), save_blocks)
+            keep = save_blocks // 2
+            for path in files[keep:]:
+                os.remove(path)
+            logger.info("deleted %d/%d block files (kept blocks 0..%d)",
+                        save_blocks - keep, save_blocks, keep - 1)
+
+            if env.hybrid:
+                vllm = env.restart_vllm(log_suffix="_p2")
+
+            # ---- Phase 2: load with holes. The request must return normally.
+            resp = send_completions(vllm.base_url(), [prompt + suffix])[0]
+            self.assertTrue(resp["choices"][0]["text"])
+
+            # The engine must survive and keep serving.
+            resp2 = send_completions(vllm.base_url(), ["engine alive?"])[0]
+            self.assertTrue(resp2["choices"][0]["text"])
+
+            failed_tasks = env.scan_connector_logs(r"load task failed")
+            self.assertTrue(failed_tasks, "no load failure was logged; the "
+                            "sabotage did not break any loaded block")
+
+            if env.hybrid:
+                # report_failures=False path: swallowed but logged.
+                swallowed = env.scan_connector_logs(
+                    r"load failed for \d+/\d+ blocks .*hybrid")
+                self.assertTrue(swallowed,
+                                "hybrid load failure was not logged")
+                return
+
+            # Full-attention: vLLM was told about the invalid blocks...
+            reported = env.scan_connector_logs(r"block_ids_with_load_errors")
+            self.assertTrue(reported, "failed loads were not reported to vLLM")
+
+            # ...and every surviving head block's loaded KV is bit-exact,
+            # while any mismatch belongs to a deleted (recomputed) block.
+            wait_for_captures(env.capture_dir, "loaded", expected=keep,
+                              timeout=180)
+            report = compare_captures(env.capture_dir, tp_size=1)
+            hashes = full_block_hashes(toks, mbs)
+            kept_keys = {("tp0", h) for h in hashes[:keep]}
+            deleted_keys = {("tp0", h) for h in hashes[keep:]}
+            failed_keys = {f["key"] for f in report["failures"]}
+            self.assertFalse(
+                failed_keys & kept_keys,
+                f"surviving loaded blocks mismatched: {failed_keys & kept_keys}")
+            self.assertTrue(
+                failed_keys <= deleted_keys,
+                f"mismatches outside the deleted blocks: "
+                f"{failed_keys - deleted_keys}")
+            matched_kept = kept_keys & set(report["matched_keys"])
+            self.assertEqual(
+                len(matched_kept), keep,
+                f"only {len(matched_kept)}/{keep} surviving blocks verified")
+        finally:
+            env.stop()
+
+
+if __name__ == "__main__":
+    unittest.main()