From 37bbc6067021278981b570fe67f1ed7ffbb76e54 Mon Sep 17 00:00:00 2001
From: FujitsuPolycom <87842395+FujitsuPolycom@users.noreply.github.com>
Date: Sun, 30 Aug 2026 02:25:28 -0500
Subject: [PATCH 1/6] Validate recurrent proof at publication time

vLLM exposes a hash-proven aligned recurrent block only after the prefill step that produces it, while nonaligned publication boundaries have no separate hand-off. Retain recurrent new-request store state through the following cached step, validate only when that step can publish, and require mappings only for recurrent groups exactly aligned at the store boundary. Nonaligned groups use the authoritative partial page in the accumulated request table; unexpected overrides and missing aligned proofs still fail closed.

Cache namespace impact: none. CacheIdentity values, digest salts, 256-token geometry, manifest schemas, page-delta bytes, and the page-tail-cow-v1 namespace are unchanged.

Validation: python -m pytest sparkcache -q (763 passed, 7 skipped); python -m pytest deploy -q (108 passed, 1 skipped); python -m ruff check .; git diff --check.
---
 deploy/deepseek_v4/tp4_profile.json         |   2 +-
 deploy/glm52_35bpw/profile.json             |   2 +-
 sparkcache/README.md                        |  18 +-
 sparkcache/spark_context_cache_connector.py |  96 ++++-----
 sparkcache/test_defect_regressions.py       | 208 +++++++++++++++++++-
 5 files changed, 262 insertions(+), 64 deletions(-)

diff --git a/deploy/deepseek_v4/tp4_profile.json b/deploy/deepseek_v4/tp4_profile.json
index c586573..8bbc09b 100644
--- a/deploy/deepseek_v4/tp4_profile.json
+++ b/deploy/deepseek_v4/tp4_profile.json
@@ -4,7 +4,7 @@
   "cache_model_profile": "deepseek-v4-fp8-hma",
   "published_runtime_base": "ghcr.io/fujitsupolycom/gb10-vllm-serving@sha256:6fc26fdad81a18f0fff67ce0a05f6d90165625ea2e1cac8a6f39bfb462017028",
   "sparkcache": {
-    "source_sha256": "83853050f790b18af95d424fec837abeb1a9a33f0538b5e4b97c16fb9c681781"
+    "source_sha256": "05f74f69e514ca5984bb6108e3bb0831efadc216b73899b1fb3cfcc29b3492ab"
   },
   "model": {
     "repository": "deepseek-ai/DeepSeek-V4-Flash-0731",
diff --git a/deploy/glm52_35bpw/profile.json b/deploy/glm52_35bpw/profile.json
index 547ad8b..df45886 100644
--- a/deploy/glm52_35bpw/profile.json
+++ b/deploy/glm52_35bpw/profile.json
@@ -5,7 +5,7 @@
   "published_runtime_base": "ghcr.io/fujitsupolycom/gb10-vllm-serving@sha256:6fc26fdad81a18f0fff67ce0a05f6d90165625ea2e1cac8a6f39bfb462017028",
   "base_image_requirement": "exact GLM-5.2 3.5-bpw R7 image recorded by the source container inspection",
   "sparkcache": {
-    "source_sha256": "83853050f790b18af95d424fec837abeb1a9a33f0538b5e4b97c16fb9c681781"
+    "source_sha256": "05f74f69e514ca5984bb6108e3bb0831efadc216b73899b1fb3cfcc29b3492ab"
   },
   "model": {
     "repository": "brandonmusic/GLM-5.2-EXL3-TR3v4-3.5bpw-MTP78",
diff --git a/sparkcache/README.md b/sparkcache/README.md
index 3998788..86dc078 100644
--- a/sparkcache/README.md
+++ b/sparkcache/README.md
@@ -225,13 +225,17 @@ when placement completes and intentionally excludes that bookkeeping.
   `sparkcache-hybrid-page-delta/v1` codec reuses only byte-identical page
   prefixes and binds the base snapshot, layout, block counts, and semantic
   token boundaries. A boundary inside an HMA page replaces that complete page
-  while retaining earlier byte-identical pages. For an aligned recurrent group,
-  vLLM may retain the replay-boundary page outside the advancing request block
-  table. Its `SchedulerOutput.recurrent_boundary_blocks` hand-off names the
-  pinned physical block by request, group, and token boundary. SparkCache uses
-  that block only after all three identities and the recurrent topology match;
-  missing or contradictory metadata skips publication rather than scanning
-  later running or speculative state. The
+  while retaining earlier byte-identical pages. At an exact recurrent-page
+  boundary, vLLM may retain the replay-boundary page outside the advancing
+  request block table. Its `SchedulerOutput.recurrent_boundary_blocks` hand-off
+  names the pinned physical block by request, group, and token boundary.
+  SparkCache defers a new recurrent request until a later cached scheduler step,
+  when the preceding forward's hand-off can be observed. It then requires one
+  matching entry for every recurrent group whose block size exactly divides the
+  publication boundary; missing or contradictory proof skips publication
+  rather than scanning later running or speculative state. At a boundary inside
+  a recurrent page, the request table's partial page remains authoritative and
+  an unexpected override is rejected. The
   `sparkcache-page-delta-manifest/v2` schema embeds its authenticated base graph
   and groups delta bytes into immutable objects of at most 64 MiB. A
   1,575,821,491-byte delta therefore uses at most 24 physical delta objects
diff --git a/sparkcache/spark_context_cache_connector.py b/sparkcache/spark_context_cache_connector.py
index 22b98fd..5c2564f 100644
--- a/sparkcache/spark_context_cache_connector.py
+++ b/sparkcache/spark_context_cache_connector.py
@@ -1616,12 +1616,13 @@ def _validated_recurrent_boundary_blocks(
     ) -> tuple[tuple[int, int], ...] | None:
         """Validate vLLM's exact recurrent replay-boundary block hand-off.
 
-        An empty tuple is valid only when the registered topology has no
-        aligned recurrent group. None means the metadata is absent,
-        incomplete, or contradictory, so publication must be skipped.
-        SparkCache never derives a replacement from another non-null table
-        entry because later entries can hold running or speculative state
-        beyond ``boundary_tokens``.
+        An empty tuple is valid when the registered topology has no recurrent
+        group exactly aligned at ``boundary_tokens``. A nonaligned recurrent
+        group's partial page remains authoritative in the request block table.
+        None means required metadata is absent, incomplete, or contradictory,
+        so publication must be skipped. SparkCache never derives an aligned
+        replacement from another non-null table entry because later entries
+        can hold running or speculative state beyond ``boundary_tokens``.
         """
 
         def reject(reason: str) -> None:
@@ -1635,20 +1636,31 @@ def reject(reason: str) -> None:
             )
             return None
 
-        required_groups = {
+        recurrent_groups = {
             group_index
             for group_index, topology in enumerate(self._group_topology)
             if topology["reuse_policy"] == "recurrent_align"
         }
-        if not required_groups:
+        if not recurrent_groups:
             return ()
+        required_groups = {
+            group_index
+            for group_index in recurrent_groups
+            if boundary_tokens
+            % int(self._group_topology[group_index]["block_size"])
+            == 0
+        }
         raw = getattr(scheduler_output, "recurrent_boundary_blocks", None)
         if raw is None:
+            if not required_groups:
+                return ()
             return reject("vLLM supplied no recurrent boundary mapping")
         if not isinstance(raw, Mapping):
             return reject("top-level value is not a mapping")
         entries = raw.get(request_id)
         if entries is None:
+            if not required_groups:
+                return ()
             return reject("request has no recurrent boundary entries")
         if not isinstance(entries, (list, tuple)):
             return reject("request value is not a sequence")
@@ -1668,6 +1680,8 @@ def reject(reason: str) -> None:
             topology = self._group_topology[group_index]
             if topology["reuse_policy"] != "recurrent_align":
                 return reject("group is not an aligned recurrent cache")
+            if group_index not in required_groups:
+                return reject("recurrent group is not aligned at the store boundary")
             if block_id <= 0:
                 return reject("physical block is vLLM's null block")
             if entry_boundary != boundary_tokens:
@@ -1741,15 +1755,6 @@ def build_connector_meta(
                 if self._has_full_quorum(digest):
                     self.counters["store_skipped_quorum"] += 1
                     continue
-                recurrent_boundary_blocks = (
-                    self._validated_recurrent_boundary_blocks(
-                        scheduler_output,
-                        req_id,
-                        span,
-                    )
-                )
-                if recurrent_boundary_blocks is None:
-                    continue
                 already = new_req.num_computed_tokens + scheduled
                 if self._streaming_snapshots_enabled:
                     self._append_streaming_snapshot_offer(
@@ -1771,10 +1776,25 @@ def build_connector_meta(
                             [list(group) for group in group_blocks],
                         )
                         self._store_token_ids[req_id] = exact_token_ids
-                        if recurrent_boundary_blocks:
-                            self._store_recurrent_boundaries[req_id] = (
-                                recurrent_boundary_blocks
-                            )
+                elif any(
+                    topology["reuse_policy"] == "recurrent_align"
+                    for topology in self._group_topology
+                ):
+                    # vLLM can only expose the hash-proven replay-boundary
+                    # block after the scheduled prefill has run. Preserve the
+                    # complete new-request table even when this step promises
+                    # the whole span; the following cached/decode step either
+                    # supplies the aligned proof or publishes the authoritative
+                    # nonaligned partial page from this table.
+                    self._store_progress[req_id] = (
+                        digest,
+                        span,
+                        already,
+                        [list(group) for group in group_blocks],
+                    )
+                    self._store_token_ids[req_id] = exact_token_ids
+                    if base_digest:
+                        self._store_bases[req_id] = (base_digest, base_span)
                 elif already >= span:
                     meta.plans.append(
                         _ReqPlan(
@@ -1787,9 +1807,6 @@ def build_connector_meta(
                             token_ids=exact_token_ids,
                             base_context_digest=base_digest,
                             base_span_tokens=base_span,
-                            recurrent_boundary_blocks=(
-                                recurrent_boundary_blocks
-                            ),
                         )
                     )
                 else:
@@ -1803,10 +1820,6 @@ def build_connector_meta(
                         [list(group) for group in group_blocks],
                     )
                     self._store_token_ids[req_id] = exact_token_ids
-                    if recurrent_boundary_blocks:
-                        self._store_recurrent_boundaries[req_id] = (
-                            recurrent_boundary_blocks
-                        )
                     if base_digest:
                         self._store_bases[req_id] = (base_digest, base_span)
         cached = scheduler_output.scheduled_cached_reqs
@@ -1816,19 +1829,6 @@ def build_connector_meta(
             digest, span, done, blocks_by_group = self._store_progress[req_id]
             exact_token_ids = self._store_token_ids.get(req_id, ())
             base_digest, base_span = self._store_bases.get(req_id, ("", 0))
-            recurrent_boundary_blocks = self._validated_recurrent_boundary_blocks(
-                scheduler_output,
-                req_id,
-                span,
-            )
-            if recurrent_boundary_blocks is None:
-                del self._store_progress[req_id]
-                self._store_token_ids.pop(req_id, None)
-                self._store_bases.pop(req_id, None)
-                self._store_recurrent_boundaries.pop(req_id, None)
-                continue
-            if recurrent_boundary_blocks:
-                self._store_recurrent_boundaries[req_id] = recurrent_boundary_blocks
             new_block_ids = cached.new_block_ids[index]
             appended = (
                 [
@@ -1882,13 +1882,19 @@ def build_connector_meta(
                     block_ids=blocks,
                 )
             elif done >= span:
+                recurrent_boundary_blocks = (
+                    self._validated_recurrent_boundary_blocks(
+                        scheduler_output,
+                        req_id,
+                        span,
+                    )
+                )
                 del self._store_progress[req_id]
                 self._store_token_ids.pop(req_id, None)
                 self._store_bases.pop(req_id, None)
-                recurrent_boundary_blocks = self._store_recurrent_boundaries.pop(
-                    req_id,
-                    (),
-                )
+                self._store_recurrent_boundaries.pop(req_id, None)
+                if recurrent_boundary_blocks is None:
+                    continue
                 if self._has_full_quorum(digest):
                     self.counters["store_skipped_quorum"] += 1
                     continue
diff --git a/sparkcache/test_defect_regressions.py b/sparkcache/test_defect_regressions.py
index 70e4bd2..a621576 100644
--- a/sparkcache/test_defect_regressions.py
+++ b/sparkcache/test_defect_regressions.py
@@ -1025,6 +1025,8 @@ class DefectD17RecurrentBoundaryMetadataTests(unittest.TestCase):
 
     BOUNDARY = 6912
     PROMPT_TOKENS = 6992
+    NONALIGNED_BOUNDARY = 8192
+    NONALIGNED_PROMPT_TOKENS = 8256
     BOUNDARY_BLOCK = 42
 
     @staticmethod
@@ -1093,6 +1095,31 @@ def _scheduler_output(
             output.recurrent_boundary_blocks = recurrent_boundary_blocks
         return output
 
+    @classmethod
+    def _cached_scheduler_output(
+        cls,
+        *,
+        num_computed_tokens: int,
+        recurrent_boundary_blocks: object = None,
+        group_count: int = 2,
+        num_scheduled_tokens: int = 1,
+    ) -> types.SimpleNamespace:
+        request_id = "dflash-recurrent-boundary"
+        output = types.SimpleNamespace(
+            scheduled_new_reqs=[],
+            scheduled_cached_reqs=types.SimpleNamespace(
+                req_ids=[request_id],
+                resumed_req_ids=set(),
+                num_computed_tokens=[num_computed_tokens],
+                new_block_ids=[tuple(() for _ in range(group_count))],
+            ),
+            num_scheduled_tokens={request_id: num_scheduled_tokens},
+            preempted_req_ids=set(),
+        )
+        if recurrent_boundary_blocks is not None:
+            output.recurrent_boundary_blocks = recurrent_boundary_blocks
+        return output
+
     def test_explicit_boundary_block_round_trips_through_manifest_store(self) -> None:
         with tempfile.TemporaryDirectory() as directory:
             root = Path(directory)
@@ -1108,12 +1135,18 @@ def test_explicit_boundary_block_round_trips_through_manifest_store(self) -> Non
                 kv_cache_config=config,
                 extra_config={"spark_cache_model_profile": "glm53-flash-hybrid"},
             )
-            output = self._scheduler_output(
-                {
+            first_metadata = scheduler.build_connector_meta(
+                self._scheduler_output()
+            )
+            self.assertEqual(first_metadata.plans, [])
+            self.assertIn("dflash-recurrent-boundary", scheduler._store_progress)
+            output = self._cached_scheduler_output(
+                num_computed_tokens=self.PROMPT_TOKENS,
+                recurrent_boundary_blocks={
                     "dflash-recurrent-boundary": [
                         (1, self.BOUNDARY_BLOCK, self.BOUNDARY)
                     ]
-                }
+                },
             )
 
             self.assertTrue(scheduler.supports_recurrent_boundary_blocks)
@@ -1173,6 +1206,139 @@ def test_explicit_boundary_block_round_trips_through_manifest_store(self) -> Non
                 torch.equal(pools["recurrent"][[93]], expected_recurrent)
             )
 
+    def test_nonaligned_boundary_uses_request_table_without_mapping(self) -> None:
+        with tempfile.TemporaryDirectory() as directory:
+            connector = _make_connector(
+                Path(directory),
+                0,
+                block_size=256,
+                role=KVConnectorRole.SCHEDULER,
+                override_worker_rank=False,
+                tp=1,
+                dcp=1,
+                kv_cache_config=self._config(),
+                extra_config={"spark_cache_model_profile": "glm53-flash-hybrid"},
+            )
+            output = self._scheduler_output()
+            request = output.scheduled_new_reqs[0]
+            request.prompt_token_ids = list(range(self.NONALIGNED_PROMPT_TOKENS))
+            output.num_scheduled_tokens[request.req_id] = (
+                self.NONALIGNED_PROMPT_TOKENS
+            )
+
+            first_metadata = connector.build_connector_meta(output)
+            self.assertEqual(first_metadata.plans, [])
+            self.assertIn(request.req_id, connector._store_progress)
+            metadata = connector.build_connector_meta(
+                self._cached_scheduler_output(
+                    num_computed_tokens=self.NONALIGNED_PROMPT_TOKENS,
+                )
+            )
+
+            self.assertEqual(len(metadata.plans), 1)
+            plan = metadata.plans[0]
+            self.assertEqual(plan.span_tokens, self.NONALIGNED_BOUNDARY)
+            self.assertEqual(plan.recurrent_boundary_blocks, ())
+            self.assertEqual(
+                connector._select_group_blocks_for_span(
+                    plan.block_ids_by_group,
+                    plan.span_tokens,
+                    recurrent_boundary_blocks=plan.recurrent_boundary_blocks,
+                ),
+                ((11, 12, 13, 14), (71,)),
+            )
+            self.assertEqual(
+                connector.counters["recurrent_boundary_metadata_rejected"],
+                0,
+            )
+
+    def test_nonaligned_boundary_rejects_unexpected_mapping(self) -> None:
+        output = self._scheduler_output()
+        request = output.scheduled_new_reqs[0]
+        request.prompt_token_ids = list(range(self.NONALIGNED_PROMPT_TOKENS))
+        output.num_scheduled_tokens[request.req_id] = self.NONALIGNED_PROMPT_TOKENS
+        with tempfile.TemporaryDirectory() as directory:
+            connector = _make_connector(
+                Path(directory),
+                0,
+                block_size=256,
+                role=KVConnectorRole.SCHEDULER,
+                override_worker_rank=False,
+                tp=1,
+                dcp=1,
+                kv_cache_config=self._config(),
+                extra_config={"spark_cache_model_profile": "glm53-flash-hybrid"},
+            )
+
+            first_metadata = connector.build_connector_meta(output)
+            self.assertEqual(first_metadata.plans, [])
+            metadata = connector.build_connector_meta(
+                self._cached_scheduler_output(
+                    num_computed_tokens=self.NONALIGNED_PROMPT_TOKENS,
+                    recurrent_boundary_blocks={
+                        "dflash-recurrent-boundary": [
+                            (
+                                1,
+                                self.BOUNDARY_BLOCK,
+                                self.NONALIGNED_BOUNDARY,
+                            )
+                        ]
+                    },
+                )
+            )
+
+            self.assertEqual(metadata.plans, [])
+            self.assertEqual(
+                connector.counters["recurrent_boundary_metadata_rejected"],
+                1,
+            )
+
+    def test_chunked_prefill_validates_only_at_publication_step(self) -> None:
+        with tempfile.TemporaryDirectory() as directory:
+            connector = _make_connector(
+                Path(directory),
+                0,
+                block_size=256,
+                role=KVConnectorRole.SCHEDULER,
+                override_worker_rank=False,
+                tp=1,
+                dcp=1,
+                kv_cache_config=self._config(),
+                extra_config={"spark_cache_model_profile": "glm53-flash-hybrid"},
+            )
+            first = self._scheduler_output()
+            first.num_scheduled_tokens["dflash-recurrent-boundary"] = 2304
+
+            self.assertEqual(connector.build_connector_meta(first).plans, [])
+            middle = self._cached_scheduler_output(
+                num_computed_tokens=2304,
+                num_scheduled_tokens=2304,
+            )
+            self.assertEqual(connector.build_connector_meta(middle).plans, [])
+            self.assertIn("dflash-recurrent-boundary", connector._store_progress)
+            self.assertEqual(
+                connector.counters["recurrent_boundary_metadata_rejected"],
+                0,
+            )
+            final = self._cached_scheduler_output(
+                num_computed_tokens=4608,
+                num_scheduled_tokens=2304,
+                recurrent_boundary_blocks={
+                    "dflash-recurrent-boundary": [
+                        (1, self.BOUNDARY_BLOCK, self.BOUNDARY)
+                    ]
+                },
+            )
+
+            metadata = connector.build_connector_meta(final)
+
+            self.assertEqual(len(metadata.plans), 1)
+            self.assertEqual(
+                metadata.plans[0].recurrent_boundary_blocks,
+                ((1, self.BOUNDARY_BLOCK),),
+            )
+            self.assertNotIn("dflash-recurrent-boundary", connector._store_progress)
+
     def test_missing_or_wrong_request_metadata_skips_publication(self) -> None:
         for boundary_metadata in (
             None,
@@ -1198,7 +1364,14 @@ def test_missing_or_wrong_request_metadata_skips_publication(self) -> None:
                     recurrent = list(recurrent)
                     recurrent[2] = 69  # stale or recycled, not boundary-proven
                     output.scheduled_new_reqs[0].block_ids = (full, tuple(recurrent))
-                    metadata = connector.build_connector_meta(output)
+                    first_metadata = connector.build_connector_meta(output)
+                    self.assertEqual(first_metadata.plans, [])
+                    metadata = connector.build_connector_meta(
+                        self._cached_scheduler_output(
+                            num_computed_tokens=self.PROMPT_TOKENS,
+                            recurrent_boundary_blocks=boundary_metadata,
+                        )
+                    )
                     self.assertEqual(metadata.plans, [])
                     self.assertEqual(
                         connector.counters[
@@ -1232,9 +1405,16 @@ def test_contradictory_boundary_metadata_skips_publication(self) -> None:
                             "spark_cache_model_profile": "glm53-flash-hybrid"
                         },
                     )
+                    first_metadata = connector.build_connector_meta(
+                        self._scheduler_output()
+                    )
+                    self.assertEqual(first_metadata.plans, [])
                     metadata = connector.build_connector_meta(
-                        self._scheduler_output(
-                            {"dflash-recurrent-boundary": entries}
+                        self._cached_scheduler_output(
+                            num_computed_tokens=self.PROMPT_TOKENS,
+                            recurrent_boundary_blocks={
+                                "dflash-recurrent-boundary": entries
+                            },
                         )
                     )
                     self.assertEqual(metadata.plans, [])
@@ -1253,9 +1433,7 @@ def test_partial_recurrent_group_coverage_skips_publication(self) -> None:
             layer_names=("recurrent-2",),
         )
         config.kv_cache_groups = (*config.kv_cache_groups, second_recurrent)
-        output = self._scheduler_output(
-            {"dflash-recurrent-boundary": [(1, 42, self.BOUNDARY)]}
-        )
+        output = self._scheduler_output()
         output.scheduled_new_reqs[0].block_ids = (
             *self._tables(),
             (0, 0, 0, 81, 82, 83, 84, 85, 86, 87, 88),
@@ -1273,7 +1451,17 @@ def test_partial_recurrent_group_coverage_skips_publication(self) -> None:
                 extra_config={"spark_cache_model_profile": "glm53-flash-hybrid"},
             )
 
-            metadata = connector.build_connector_meta(output)
+            first_metadata = connector.build_connector_meta(output)
+            self.assertEqual(first_metadata.plans, [])
+            metadata = connector.build_connector_meta(
+                self._cached_scheduler_output(
+                    num_computed_tokens=self.PROMPT_TOKENS,
+                    recurrent_boundary_blocks={
+                        "dflash-recurrent-boundary": [(1, 42, self.BOUNDARY)]
+                    },
+                    group_count=3,
+                )
+            )
 
             self.assertEqual(metadata.plans, [])
             self.assertEqual(

From 99e101ddbc62d24f853c9210cf1b1a164c4c265a Mon Sep 17 00:00:00 2001
From: FujitsuPolycom <87842395+FujitsuPolycom@users.noreply.github.com>
Date: Sun, 30 Aug 2026 02:34:18 -0500
Subject: [PATCH 2/6] Latch every proven recurrent CoW target

A recurrent partial page can be replaced after the initial request table is observed. Its durable publication source is the pinned block delivered by vLLM partial_tail_offloads, not the accumulated source ID. Treat absent per-request metadata as pending, latch complete validated mappings from any scheduler output, reject incomplete or conflicting evidence, and publish only after every recurrent group has a proven block. Preemption clears the latch; completion and quorum retire pending state.

Cache namespace impact: none. CacheIdentity values, digest salts, 256-token geometry, manifest schemas, page-delta bytes, and page-tail-cow-v1 are unchanged.

Validation: python -m pytest sparkcache -q (764 passed, 7 skipped after isolated timing rerun); python -m pytest deploy -q (108 passed, 1 skipped); python -m ruff check .; git diff --check.
---
 deploy/deepseek_v4/tp4_profile.json         |   2 +-
 deploy/glm52_35bpw/profile.json             |   2 +-
 sparkcache/README.md                        |  13 +-
 sparkcache/spark_context_cache_connector.py | 163 +++++++++++---------
 sparkcache/test_defect_regressions.py       | 159 +++++++++++++++----
 5 files changed, 234 insertions(+), 105 deletions(-)

diff --git a/deploy/deepseek_v4/tp4_profile.json b/deploy/deepseek_v4/tp4_profile.json
index 8bbc09b..ce63c32 100644
--- a/deploy/deepseek_v4/tp4_profile.json
+++ b/deploy/deepseek_v4/tp4_profile.json
@@ -4,7 +4,7 @@
   "cache_model_profile": "deepseek-v4-fp8-hma",
   "published_runtime_base": "ghcr.io/fujitsupolycom/gb10-vllm-serving@sha256:6fc26fdad81a18f0fff67ce0a05f6d90165625ea2e1cac8a6f39bfb462017028",
   "sparkcache": {
-    "source_sha256": "05f74f69e514ca5984bb6108e3bb0831efadc216b73899b1fb3cfcc29b3492ab"
+    "source_sha256": "155a06101524d4c2d2f55dbbd01576e35d5c729888e216fd2f3963e275949ba0"
   },
   "model": {
     "repository": "deepseek-ai/DeepSeek-V4-Flash-0731",
diff --git a/deploy/glm52_35bpw/profile.json b/deploy/glm52_35bpw/profile.json
index df45886..26876f5 100644
--- a/deploy/glm52_35bpw/profile.json
+++ b/deploy/glm52_35bpw/profile.json
@@ -5,7 +5,7 @@
   "published_runtime_base": "ghcr.io/fujitsupolycom/gb10-vllm-serving@sha256:6fc26fdad81a18f0fff67ce0a05f6d90165625ea2e1cac8a6f39bfb462017028",
   "base_image_requirement": "exact GLM-5.2 3.5-bpw R7 image recorded by the source container inspection",
   "sparkcache": {
-    "source_sha256": "05f74f69e514ca5984bb6108e3bb0831efadc216b73899b1fb3cfcc29b3492ab"
+    "source_sha256": "155a06101524d4c2d2f55dbbd01576e35d5c729888e216fd2f3963e275949ba0"
   },
   "model": {
     "repository": "brandonmusic/GLM-5.2-EXL3-TR3v4-3.5bpw-MTP78",
diff --git a/sparkcache/README.md b/sparkcache/README.md
index 86dc078..3b47425 100644
--- a/sparkcache/README.md
+++ b/sparkcache/README.md
@@ -230,12 +230,13 @@ when placement completes and intentionally excludes that bookkeeping.
   request block table. Its `SchedulerOutput.recurrent_boundary_blocks` hand-off
   names the pinned physical block by request, group, and token boundary.
   SparkCache defers a new recurrent request until a later cached scheduler step,
-  when the preceding forward's hand-off can be observed. It then requires one
-  matching entry for every recurrent group whose block size exactly divides the
-  publication boundary; missing or contradictory proof skips publication
-  rather than scanning later running or speculative state. At a boundary inside
-  a recurrent page, the request table's partial page remains authoritative and
-  an unexpected override is rejected. The
+  when the preceding forward's hand-off can be observed. It latches one matching
+  entry for every recurrent group, including a partial-tail CoW target when the
+  boundary lies inside a recurrent page. Outputs with no entry leave publication
+  pending; incomplete, contradictory, or changed evidence cancels it. A store is
+  emitted only after every recurrent group has a proven pinned block. SparkCache
+  never substitutes an accumulated request-table ID because vLLM may have
+  replaced that source block while producing the durable CoW target. The
   `sparkcache-page-delta-manifest/v2` schema embeds its authenticated base graph
   and groups delta bytes into immutable objects of at most 64 MiB. A
   1,575,821,491-byte delta therefore uses at most 24 physical delta objects
diff --git a/sparkcache/spark_context_cache_connector.py b/sparkcache/spark_context_cache_connector.py
index 5c2564f..8cf5708 100644
--- a/sparkcache/spark_context_cache_connector.py
+++ b/sparkcache/spark_context_cache_connector.py
@@ -570,6 +570,11 @@ def __init__(
         self._storage_mode = config.storage_mode
         self._publication_schema = config.publication_schema
         self._group_topology = config.group_topology
+        self._recurrent_group_indexes = frozenset(
+            group_index
+            for group_index, topology in enumerate(self._group_topology)
+            if topology["reuse_policy"] == "recurrent_align"
+        )
         self._chunk_tokens = config.chunk_tokens
         self._root = config.root
         self._store = ManifestStore(self._root)
@@ -1613,16 +1618,19 @@ def _validated_recurrent_boundary_blocks(
         scheduler_output: "SchedulerOutput",
         request_id: str,
         boundary_tokens: int,
+        *,
+        latched: tuple[tuple[int, int], ...] = (),
     ) -> tuple[tuple[int, int], ...] | None:
         """Validate vLLM's exact recurrent replay-boundary block hand-off.
 
-        An empty tuple is valid when the registered topology has no recurrent
-        group exactly aligned at ``boundary_tokens``. A nonaligned recurrent
-        group's partial page remains authoritative in the request block table.
-        None means required metadata is absent, incomplete, or contradictory,
-        so publication must be skipped. SparkCache never derives an aligned
-        replacement from another non-null table entry because later entries
-        can hold running or speculative state beyond ``boundary_tokens``.
+        vLLM may expose a full-page boundary or a partial-tail CoW target on a
+        later scheduler output than the request which began the store. Absent
+        per-request metadata therefore preserves ``latched`` and keeps the
+        store pending. None means supplied metadata is incomplete,
+        contradictory, or conflicts with an earlier latch, so this publication
+        attempt must be poisoned. SparkCache never derives a replacement from
+        another request-table entry because it can name overwritten running or
+        speculative state instead of vLLM's pinned CoW target.
         """
 
         def reject(reason: str) -> None:
@@ -1636,32 +1644,17 @@ def reject(reason: str) -> None:
             )
             return None
 
-        recurrent_groups = {
-            group_index
-            for group_index, topology in enumerate(self._group_topology)
-            if topology["reuse_policy"] == "recurrent_align"
-        }
-        if not recurrent_groups:
+        required_groups = self._recurrent_group_indexes
+        if not required_groups:
             return ()
-        required_groups = {
-            group_index
-            for group_index in recurrent_groups
-            if boundary_tokens
-            % int(self._group_topology[group_index]["block_size"])
-            == 0
-        }
         raw = getattr(scheduler_output, "recurrent_boundary_blocks", None)
         if raw is None:
-            if not required_groups:
-                return ()
-            return reject("vLLM supplied no recurrent boundary mapping")
+            return latched
         if not isinstance(raw, Mapping):
             return reject("top-level value is not a mapping")
         entries = raw.get(request_id)
         if entries is None:
-            if not required_groups:
-                return ()
-            return reject("request has no recurrent boundary entries")
+            return latched
         if not isinstance(entries, (list, tuple)):
             return reject("request value is not a sequence")
 
@@ -1680,8 +1673,6 @@ def reject(reason: str) -> None:
             topology = self._group_topology[group_index]
             if topology["reuse_policy"] != "recurrent_align":
                 return reject("group is not an aligned recurrent cache")
-            if group_index not in required_groups:
-                return reject("recurrent group is not aligned at the store boundary")
             if block_id <= 0:
                 return reject("physical block is vLLM's null block")
             if entry_boundary != boundary_tokens:
@@ -1690,7 +1681,10 @@ def reject(reason: str) -> None:
             overrides.append((group_index, block_id))
         if seen_groups != required_groups:
             return reject("entries do not cover every aligned recurrent group")
-        return tuple(sorted(overrides))
+        validated = tuple(sorted(overrides))
+        if latched and validated != latched:
+            return reject("entries conflict with the latched recurrent boundary")
+        return validated
 
     def build_connector_meta(
         self, scheduler_output: "SchedulerOutput"
@@ -1776,16 +1770,21 @@ def build_connector_meta(
                             [list(group) for group in group_blocks],
                         )
                         self._store_token_ids[req_id] = exact_token_ids
-                elif any(
-                    topology["reuse_policy"] == "recurrent_align"
-                    for topology in self._group_topology
-                ):
-                    # vLLM can only expose the hash-proven replay-boundary
-                    # block after the scheduled prefill has run. Preserve the
-                    # complete new-request table even when this step promises
-                    # the whole span; the following cached/decode step either
-                    # supplies the aligned proof or publishes the authoritative
-                    # nonaligned partial page from this table.
+                elif self._recurrent_group_indexes:
+                    recurrent_boundary_blocks = (
+                        self._validated_recurrent_boundary_blocks(
+                            scheduler_output,
+                            req_id,
+                            span,
+                        )
+                    )
+                    if recurrent_boundary_blocks is None:
+                        continue
+                    # Full-page proof and partial-tail CoW hand-offs can arrive
+                    # after the prefill which began this store. Retain the
+                    # complete request table and any early proof until a later
+                    # cached step has both finished the span and proven every
+                    # recurrent group.
                     self._store_progress[req_id] = (
                         digest,
                         span,
@@ -1793,6 +1792,10 @@ def build_connector_meta(
                         [list(group) for group in group_blocks],
                     )
                     self._store_token_ids[req_id] = exact_token_ids
+                    if recurrent_boundary_blocks:
+                        self._store_recurrent_boundaries[req_id] = (
+                            recurrent_boundary_blocks
+                        )
                     if base_digest:
                         self._store_bases[req_id] = (base_digest, base_span)
                 elif already >= span:
@@ -1829,30 +1832,52 @@ def build_connector_meta(
             digest, span, done, blocks_by_group = self._store_progress[req_id]
             exact_token_ids = self._store_token_ids.get(req_id, ())
             base_digest, base_span = self._store_bases.get(req_id, ("", 0))
-            new_block_ids = cached.new_block_ids[index]
-            appended = (
-                [
-                    list(group)
-                    for group in self._normalize_group_blocks(
-                        new_block_ids,
-                        allow_empty_groups=True,
-                    )
-                ]
-                if new_block_ids is not None
-                else [[] for _ in blocks_by_group]
+            if self._has_full_quorum(digest):
+                del self._store_progress[req_id]
+                self._store_token_ids.pop(req_id, None)
+                self._store_bases.pop(req_id, None)
+                self._store_recurrent_boundaries.pop(req_id, None)
+                self.counters["store_skipped_quorum"] += 1
+                continue
+            recurrent_boundary_blocks = self._validated_recurrent_boundary_blocks(
+                scheduler_output,
+                req_id,
+                span,
+                latched=self._store_recurrent_boundaries.get(req_id, ()),
             )
-            if len(appended) != len(blocks_by_group):
-                raise RuntimeError(
-                    "spark-context-cache: KV-cache group count changed while"
-                    " accumulating a store"
+            if recurrent_boundary_blocks is None:
+                del self._store_progress[req_id]
+                self._store_token_ids.pop(req_id, None)
+                self._store_bases.pop(req_id, None)
+                self._store_recurrent_boundaries.pop(req_id, None)
+                continue
+            if recurrent_boundary_blocks:
+                self._store_recurrent_boundaries[req_id] = recurrent_boundary_blocks
+            if done < span or req_id in cached.resumed_req_ids:
+                new_block_ids = cached.new_block_ids[index]
+                appended = (
+                    [
+                        list(group)
+                        for group in self._normalize_group_blocks(
+                            new_block_ids,
+                            allow_empty_groups=True,
+                        )
+                    ]
+                    if new_block_ids is not None
+                    else [[] for _ in blocks_by_group]
                 )
-            if req_id in cached.resumed_req_ids:
-                blocks_by_group = appended
-            else:
-                blocks_by_group = [
-                    existing + added
-                    for existing, added in zip(blocks_by_group, appended)
-                ]
+                if len(appended) != len(blocks_by_group):
+                    raise RuntimeError(
+                        "spark-context-cache: KV-cache group count changed while"
+                        " accumulating a store"
+                    )
+                if req_id in cached.resumed_req_ids:
+                    blocks_by_group = appended
+                else:
+                    blocks_by_group = [
+                        existing + added
+                        for existing, added in zip(blocks_by_group, appended)
+                    ]
             blocks = blocks_by_group[0]
             done = cached.num_computed_tokens[index] + (
                 scheduler_output.num_scheduled_tokens.get(req_id, 0)
@@ -1882,22 +1907,18 @@ def build_connector_meta(
                     block_ids=blocks,
                 )
             elif done >= span:
-                recurrent_boundary_blocks = (
-                    self._validated_recurrent_boundary_blocks(
-                        scheduler_output,
-                        req_id,
+                if self._recurrent_group_indexes and not recurrent_boundary_blocks:
+                    self._store_progress[req_id] = (
+                        digest,
                         span,
+                        done,
+                        blocks_by_group,
                     )
-                )
+                    continue
                 del self._store_progress[req_id]
                 self._store_token_ids.pop(req_id, None)
                 self._store_bases.pop(req_id, None)
                 self._store_recurrent_boundaries.pop(req_id, None)
-                if recurrent_boundary_blocks is None:
-                    continue
-                if self._has_full_quorum(digest):
-                    self.counters["store_skipped_quorum"] += 1
-                    continue
                 normalized = tuple(tuple(group) for group in blocks_by_group)
                 meta.plans.append(
                     _ReqPlan(
diff --git a/sparkcache/test_defect_regressions.py b/sparkcache/test_defect_regressions.py
index a621576..d5a53ae 100644
--- a/sparkcache/test_defect_regressions.py
+++ b/sparkcache/test_defect_regressions.py
@@ -1028,6 +1028,7 @@ class DefectD17RecurrentBoundaryMetadataTests(unittest.TestCase):
     NONALIGNED_BOUNDARY = 8192
     NONALIGNED_PROMPT_TOKENS = 8256
     BOUNDARY_BLOCK = 42
+    COW_BLOCK = 142
 
     @staticmethod
     def _config() -> types.SimpleNamespace:
@@ -1103,15 +1104,21 @@ def _cached_scheduler_output(
         recurrent_boundary_blocks: object = None,
         group_count: int = 2,
         num_scheduled_tokens: int = 1,
+        resumed: bool = False,
+        new_block_ids: object = None,
     ) -> types.SimpleNamespace:
         request_id = "dflash-recurrent-boundary"
         output = types.SimpleNamespace(
             scheduled_new_reqs=[],
             scheduled_cached_reqs=types.SimpleNamespace(
                 req_ids=[request_id],
-                resumed_req_ids=set(),
+                resumed_req_ids={request_id} if resumed else set(),
                 num_computed_tokens=[num_computed_tokens],
-                new_block_ids=[tuple(() for _ in range(group_count))],
+                new_block_ids=[
+                    new_block_ids
+                    if new_block_ids is not None
+                    else tuple(() for _ in range(group_count))
+                ],
             ),
             num_scheduled_tokens={request_id: num_scheduled_tokens},
             preempted_req_ids=set(),
@@ -1135,18 +1142,24 @@ def test_explicit_boundary_block_round_trips_through_manifest_store(self) -> Non
                 kv_cache_config=config,
                 extra_config={"spark_cache_model_profile": "glm53-flash-hybrid"},
             )
+            boundary_metadata = {
+                "dflash-recurrent-boundary": [
+                    (1, self.BOUNDARY_BLOCK, self.BOUNDARY)
+                ]
+            }
             first_metadata = scheduler.build_connector_meta(
-                self._scheduler_output()
+                self._scheduler_output(boundary_metadata)
             )
             self.assertEqual(first_metadata.plans, [])
             self.assertIn("dflash-recurrent-boundary", scheduler._store_progress)
+            self.assertEqual(
+                scheduler._store_recurrent_boundaries[
+                    "dflash-recurrent-boundary"
+                ],
+                ((1, self.BOUNDARY_BLOCK),),
+            )
             output = self._cached_scheduler_output(
                 num_computed_tokens=self.PROMPT_TOKENS,
-                recurrent_boundary_blocks={
-                    "dflash-recurrent-boundary": [
-                        (1, self.BOUNDARY_BLOCK, self.BOUNDARY)
-                    ]
-                },
             )
 
             self.assertTrue(scheduler.supports_recurrent_boundary_blocks)
@@ -1206,7 +1219,7 @@ def test_explicit_boundary_block_round_trips_through_manifest_store(self) -> Non
                 torch.equal(pools["recurrent"][[93]], expected_recurrent)
             )
 
-    def test_nonaligned_boundary_uses_request_table_without_mapping(self) -> None:
+    def test_nonaligned_boundary_waits_for_partial_tail_cow_mapping(self) -> None:
         with tempfile.TemporaryDirectory() as directory:
             connector = _make_connector(
                 Path(directory),
@@ -1229,34 +1242,49 @@ def test_nonaligned_boundary_uses_request_table_without_mapping(self) -> None:
             first_metadata = connector.build_connector_meta(output)
             self.assertEqual(first_metadata.plans, [])
             self.assertIn(request.req_id, connector._store_progress)
-            metadata = connector.build_connector_meta(
+            pending = connector.build_connector_meta(
                 self._cached_scheduler_output(
                     num_computed_tokens=self.NONALIGNED_PROMPT_TOKENS,
                 )
             )
+            self.assertEqual(pending.plans, [])
+            self.assertIn(request.req_id, connector._store_progress)
+            metadata = connector.build_connector_meta(
+                self._cached_scheduler_output(
+                    num_computed_tokens=self.NONALIGNED_PROMPT_TOKENS + 1,
+                    recurrent_boundary_blocks={
+                        request.req_id: [
+                            (1, self.COW_BLOCK, self.NONALIGNED_BOUNDARY)
+                        ]
+                    },
+                )
+            )
 
             self.assertEqual(len(metadata.plans), 1)
             plan = metadata.plans[0]
             self.assertEqual(plan.span_tokens, self.NONALIGNED_BOUNDARY)
-            self.assertEqual(plan.recurrent_boundary_blocks, ())
+            self.assertEqual(plan.recurrent_boundary_blocks, ((1, self.COW_BLOCK),))
             self.assertEqual(
                 connector._select_group_blocks_for_span(
                     plan.block_ids_by_group,
                     plan.span_tokens,
                     recurrent_boundary_blocks=plan.recurrent_boundary_blocks,
                 ),
-                ((11, 12, 13, 14), (71,)),
+                ((11, 12, 13, 14), (self.COW_BLOCK,)),
             )
             self.assertEqual(
                 connector.counters["recurrent_boundary_metadata_rejected"],
                 0,
             )
 
-    def test_nonaligned_boundary_rejects_unexpected_mapping(self) -> None:
-        output = self._scheduler_output()
-        request = output.scheduled_new_reqs[0]
-        request.prompt_token_ids = list(range(self.NONALIGNED_PROMPT_TOKENS))
-        output.num_scheduled_tokens[request.req_id] = self.NONALIGNED_PROMPT_TOKENS
+    def test_conflicting_mapping_poisons_latched_publication(self) -> None:
+        output = self._scheduler_output(
+            {
+                "dflash-recurrent-boundary": [
+                    (1, self.BOUNDARY_BLOCK, self.BOUNDARY)
+                ]
+            }
+        )
         with tempfile.TemporaryDirectory() as directory:
             connector = _make_connector(
                 Path(directory),
@@ -1277,11 +1305,7 @@ def test_nonaligned_boundary_rejects_unexpected_mapping(self) -> None:
                     num_computed_tokens=self.NONALIGNED_PROMPT_TOKENS,
                     recurrent_boundary_blocks={
                         "dflash-recurrent-boundary": [
-                            (
-                                1,
-                                self.BOUNDARY_BLOCK,
-                                self.NONALIGNED_BOUNDARY,
-                            )
+                            (1, self.BOUNDARY_BLOCK + 1, self.BOUNDARY)
                         ]
                     },
                 )
@@ -1339,7 +1363,7 @@ def test_chunked_prefill_validates_only_at_publication_step(self) -> None:
             )
             self.assertNotIn("dflash-recurrent-boundary", connector._store_progress)
 
-    def test_missing_or_wrong_request_metadata_skips_publication(self) -> None:
+    def test_missing_or_wrong_request_metadata_keeps_publication_pending(self) -> None:
         for boundary_metadata in (
             None,
             {"another-request": [(1, self.BOUNDARY_BLOCK, self.BOUNDARY)]},
@@ -1377,7 +1401,23 @@ def test_missing_or_wrong_request_metadata_skips_publication(self) -> None:
                         connector.counters[
                             "recurrent_boundary_metadata_rejected"
                         ],
-                        1,
+                        0,
+                    )
+                    self.assertIn(
+                        "dflash-recurrent-boundary", connector._store_progress
+                    )
+                    connector.request_finished(
+                        types.SimpleNamespace(
+                            request_id="dflash-recurrent-boundary"
+                        ),
+                        [],
+                    )
+                    self.assertNotIn(
+                        "dflash-recurrent-boundary", connector._store_progress
+                    )
+                    self.assertNotIn(
+                        "dflash-recurrent-boundary",
+                        connector._store_recurrent_boundaries,
                     )
 
     def test_contradictory_boundary_metadata_skips_publication(self) -> None:
@@ -1483,9 +1523,16 @@ def test_preemption_discards_request_lifetime_boundary_block(self) -> None:
                 extra_config={"spark_cache_model_profile": "glm53-flash-hybrid"},
             )
             request_id = "dflash-recurrent-boundary"
-            connector._store_recurrent_boundaries[request_id] = (
-                (1, self.BOUNDARY_BLOCK),
+            connector.build_connector_meta(
+                self._scheduler_output(
+                    {
+                        request_id: [
+                            (1, self.BOUNDARY_BLOCK, self.BOUNDARY)
+                        ]
+                    }
+                )
             )
+            self.assertIn(request_id, connector._store_recurrent_boundaries)
             output = types.SimpleNamespace(
                 scheduled_new_reqs=[],
                 scheduled_cached_reqs=types.SimpleNamespace(
@@ -1502,6 +1549,66 @@ def test_preemption_discards_request_lifetime_boundary_block(self) -> None:
 
             self.assertEqual(metadata.preempted_request_ids, (request_id,))
             self.assertNotIn(request_id, connector._store_recurrent_boundaries)
+            self.assertIn(request_id, connector._store_progress)
+
+            resumed = connector.build_connector_meta(
+                self._cached_scheduler_output(
+                    num_computed_tokens=self.PROMPT_TOKENS,
+                    resumed=True,
+                    new_block_ids=self._tables(),
+                )
+            )
+            self.assertEqual(resumed.plans, [])
+            self.assertNotIn(request_id, connector._store_recurrent_boundaries)
+            published = connector.build_connector_meta(
+                self._cached_scheduler_output(
+                    num_computed_tokens=self.PROMPT_TOKENS + 1,
+                    recurrent_boundary_blocks={
+                        request_id: [
+                            (1, self.BOUNDARY_BLOCK + 10, self.BOUNDARY)
+                        ]
+                    },
+                )
+            )
+            self.assertEqual(len(published.plans), 1)
+            self.assertEqual(
+                published.plans[0].recurrent_boundary_blocks,
+                ((1, self.BOUNDARY_BLOCK + 10),),
+            )
+
+    def test_quorum_retires_pending_store_without_boundary_proof(self) -> None:
+        with tempfile.TemporaryDirectory() as directory:
+            connector = _make_connector(
+                Path(directory),
+                0,
+                block_size=256,
+                role=KVConnectorRole.SCHEDULER,
+                override_worker_rank=False,
+                tp=1,
+                dcp=1,
+                kv_cache_config=self._config(),
+                extra_config={"spark_cache_model_profile": "glm53-flash-hybrid"},
+            )
+            request_id = "dflash-recurrent-boundary"
+            connector.build_connector_meta(self._scheduler_output())
+            digest = connector._store_progress[request_id][0]
+            connector._quorum[digest] = {0}
+
+            metadata = connector.build_connector_meta(
+                self._cached_scheduler_output(
+                    num_computed_tokens=self.PROMPT_TOKENS,
+                    recurrent_boundary_blocks="malformed-but-unneeded",
+                )
+            )
+
+            self.assertEqual(metadata.plans, [])
+            self.assertNotIn(request_id, connector._store_progress)
+            self.assertNotIn(request_id, connector._store_recurrent_boundaries)
+            self.assertEqual(connector.counters["store_skipped_quorum"], 1)
+            self.assertEqual(
+                connector.counters["recurrent_boundary_metadata_rejected"],
+                0,
+            )
 
 
 class DigestNamespaceTests(unittest.TestCase):

From 972b203a716eb20f1889583f7f408788f2a67684 Mon Sep 17 00:00:00 2001
From: FujitsuPolycom <87842395+FujitsuPolycom@users.noreply.github.com>
Date: Sun, 30 Aug 2026 02:37:22 -0500
Subject: [PATCH 3/6] Accept the boundary-crossing producer postimage

The recurrent producer now identifies a replay-boundary block when one cache_blocks call crosses that boundary, rather than requiring the caller's token count to equal it. Advance the exact single_type_kv_cache_manager postimage and lease-contract receipt while retaining the same required symbol surface.

Cache namespace impact: none. CacheIdentity values, digest salts, chunk geometry, manifest schemas, page-delta bytes, and page-tail-cow-v1 are unchanged.

Validation: strict eleven-file verifier passed against the composed vLLM source; SparkCache 764 passed, 7 skipped; deploy 108 passed, 1 skipped; Ruff and diff checks passed.
---
 deploy/deepseek_v4/tp4_profile.json                             | 2 +-
 deploy/glm52_35bpw/profile.json                                 | 2 +-
 patches/vllm-glm53-b12x-kda-adaptive-mtp/source-receipt.json    | 2 +-
 ...llm-kv-block-lease-contract-glm53-b12x-kda-adaptive-mtp.json | 2 +-
 4 files changed, 4 insertions(+), 4 deletions(-)

diff --git a/deploy/deepseek_v4/tp4_profile.json b/deploy/deepseek_v4/tp4_profile.json
index ce63c32..4535e95 100644
--- a/deploy/deepseek_v4/tp4_profile.json
+++ b/deploy/deepseek_v4/tp4_profile.json
@@ -4,7 +4,7 @@
   "cache_model_profile": "deepseek-v4-fp8-hma",
   "published_runtime_base": "ghcr.io/fujitsupolycom/gb10-vllm-serving@sha256:6fc26fdad81a18f0fff67ce0a05f6d90165625ea2e1cac8a6f39bfb462017028",
   "sparkcache": {
-    "source_sha256": "155a06101524d4c2d2f55dbbd01576e35d5c729888e216fd2f3963e275949ba0"
+    "source_sha256": "0c7547fb7e78b3af202d83690170efec2c7602a7c7ea6b407ef70c3fcdd8cfbb"
   },
   "model": {
     "repository": "deepseek-ai/DeepSeek-V4-Flash-0731",
diff --git a/deploy/glm52_35bpw/profile.json b/deploy/glm52_35bpw/profile.json
index 26876f5..9c63215 100644
--- a/deploy/glm52_35bpw/profile.json
+++ b/deploy/glm52_35bpw/profile.json
@@ -5,7 +5,7 @@
   "published_runtime_base": "ghcr.io/fujitsupolycom/gb10-vllm-serving@sha256:6fc26fdad81a18f0fff67ce0a05f6d90165625ea2e1cac8a6f39bfb462017028",
   "base_image_requirement": "exact GLM-5.2 3.5-bpw R7 image recorded by the source container inspection",
   "sparkcache": {
-    "source_sha256": "155a06101524d4c2d2f55dbbd01576e35d5c729888e216fd2f3963e275949ba0"
+    "source_sha256": "0c7547fb7e78b3af202d83690170efec2c7602a7c7ea6b407ef70c3fcdd8cfbb"
   },
   "model": {
     "repository": "brandonmusic/GLM-5.2-EXL3-TR3v4-3.5bpw-MTP78",
diff --git a/patches/vllm-glm53-b12x-kda-adaptive-mtp/source-receipt.json b/patches/vllm-glm53-b12x-kda-adaptive-mtp/source-receipt.json
index f4a4d28..abb2cae 100644
--- a/patches/vllm-glm53-b12x-kda-adaptive-mtp/source-receipt.json
+++ b/patches/vllm-glm53-b12x-kda-adaptive-mtp/source-receipt.json
@@ -62,7 +62,7 @@
   ],
   "contract": {
     "path": "sparkcache/runtime_patches/vllm-kv-block-lease-contract-glm53-b12x-kda-adaptive-mtp.json",
-    "sha256": "f36ed14eaf1f97a5dffa94bda8151b1e0fa182afc0d121b757b70bebc6a43811"
+    "sha256": "70b94520f1094d99ebf0e2a3f5a61e29ca377f61f6bd052bd899f23d034957b8"
   },
   "result": "All four SparkCache patches apply in order to the LF source tree and produce the exact preimages required by the recurrent-boundary producer. The eleven-file final-runtime contract verifies only after that producer creates its four recurrent postimages; it also covers the unchanged live-tensor B12X KDA surface."
 }
diff --git a/sparkcache/runtime_patches/vllm-kv-block-lease-contract-glm53-b12x-kda-adaptive-mtp.json b/sparkcache/runtime_patches/vllm-kv-block-lease-contract-glm53-b12x-kda-adaptive-mtp.json
index 7de6ead..39c83ba 100644
--- a/sparkcache/runtime_patches/vllm-kv-block-lease-contract-glm53-b12x-kda-adaptive-mtp.json
+++ b/sparkcache/runtime_patches/vllm-kv-block-lease-contract-glm53-b12x-kda-adaptive-mtp.json
@@ -100,7 +100,7 @@
     {
     "path": "vllm/v1/core/single_type_kv_cache_manager.py",
     "accepted_sha256": {
-      "recurrent_boundary_contract": "f67a1850a7e0288baaa6d42e7ec55b22b09c156720767e23acaabedcae333c8a"
+      "recurrent_boundary_contract": "2ab95dea008d65488bc2d55ccbe023c4481dba633fe92924001b9e4155ff38a2"
       },
       "required_symbols": [
         "SingleTypeKVCacheManager.add_local_computed_blocks",

From bf7174e341e032d9b5cc970cca3d6c2985d364fc Mon Sep 17 00:00:00 2001
From: FujitsuPolycom <87842395+FujitsuPolycom@users.noreply.github.com>
Date: Sun, 30 Aug 2026 02:54:38 -0500
Subject: [PATCH 4/6] Retain the exact-stop recurrent producer contract

An overshooting Mamba cache_blocks call has already nulled the earlier arithmetic boundary slots, so the crossing postimage cannot prove or recover that state. Restore the verified exact-stop producer postimage and lease contract. Scheduler-level regression coverage owns the invariant that aligned GLM prefill stops at the 2,304-token boundary; nonaligned 8,192 publication uses the next-step partial-tail CoW hand-off latched by SparkCache.

Cache namespace impact: none. CacheIdentity values, digest salts, chunk geometry, manifest schemas, page-delta bytes, and page-tail-cow-v1 are unchanged.
---
 deploy/deepseek_v4/tp4_profile.json                             | 2 +-
 deploy/glm52_35bpw/profile.json                                 | 2 +-
 patches/vllm-glm53-b12x-kda-adaptive-mtp/source-receipt.json    | 2 +-
 ...llm-kv-block-lease-contract-glm53-b12x-kda-adaptive-mtp.json | 2 +-
 4 files changed, 4 insertions(+), 4 deletions(-)

diff --git a/deploy/deepseek_v4/tp4_profile.json b/deploy/deepseek_v4/tp4_profile.json
index 4535e95..ce63c32 100644
--- a/deploy/deepseek_v4/tp4_profile.json
+++ b/deploy/deepseek_v4/tp4_profile.json
@@ -4,7 +4,7 @@
   "cache_model_profile": "deepseek-v4-fp8-hma",
   "published_runtime_base": "ghcr.io/fujitsupolycom/gb10-vllm-serving@sha256:6fc26fdad81a18f0fff67ce0a05f6d90165625ea2e1cac8a6f39bfb462017028",
   "sparkcache": {
-    "source_sha256": "0c7547fb7e78b3af202d83690170efec2c7602a7c7ea6b407ef70c3fcdd8cfbb"
+    "source_sha256": "155a06101524d4c2d2f55dbbd01576e35d5c729888e216fd2f3963e275949ba0"
   },
   "model": {
     "repository": "deepseek-ai/DeepSeek-V4-Flash-0731",
diff --git a/deploy/glm52_35bpw/profile.json b/deploy/glm52_35bpw/profile.json
index 9c63215..26876f5 100644
--- a/deploy/glm52_35bpw/profile.json
+++ b/deploy/glm52_35bpw/profile.json
@@ -5,7 +5,7 @@
   "published_runtime_base": "ghcr.io/fujitsupolycom/gb10-vllm-serving@sha256:6fc26fdad81a18f0fff67ce0a05f6d90165625ea2e1cac8a6f39bfb462017028",
   "base_image_requirement": "exact GLM-5.2 3.5-bpw R7 image recorded by the source container inspection",
   "sparkcache": {
-    "source_sha256": "0c7547fb7e78b3af202d83690170efec2c7602a7c7ea6b407ef70c3fcdd8cfbb"
+    "source_sha256": "155a06101524d4c2d2f55dbbd01576e35d5c729888e216fd2f3963e275949ba0"
   },
   "model": {
     "repository": "brandonmusic/GLM-5.2-EXL3-TR3v4-3.5bpw-MTP78",
diff --git a/patches/vllm-glm53-b12x-kda-adaptive-mtp/source-receipt.json b/patches/vllm-glm53-b12x-kda-adaptive-mtp/source-receipt.json
index abb2cae..f4a4d28 100644
--- a/patches/vllm-glm53-b12x-kda-adaptive-mtp/source-receipt.json
+++ b/patches/vllm-glm53-b12x-kda-adaptive-mtp/source-receipt.json
@@ -62,7 +62,7 @@
   ],
   "contract": {
     "path": "sparkcache/runtime_patches/vllm-kv-block-lease-contract-glm53-b12x-kda-adaptive-mtp.json",
-    "sha256": "70b94520f1094d99ebf0e2a3f5a61e29ca377f61f6bd052bd899f23d034957b8"
+    "sha256": "f36ed14eaf1f97a5dffa94bda8151b1e0fa182afc0d121b757b70bebc6a43811"
   },
   "result": "All four SparkCache patches apply in order to the LF source tree and produce the exact preimages required by the recurrent-boundary producer. The eleven-file final-runtime contract verifies only after that producer creates its four recurrent postimages; it also covers the unchanged live-tensor B12X KDA surface."
 }
diff --git a/sparkcache/runtime_patches/vllm-kv-block-lease-contract-glm53-b12x-kda-adaptive-mtp.json b/sparkcache/runtime_patches/vllm-kv-block-lease-contract-glm53-b12x-kda-adaptive-mtp.json
index 39c83ba..7de6ead 100644
--- a/sparkcache/runtime_patches/vllm-kv-block-lease-contract-glm53-b12x-kda-adaptive-mtp.json
+++ b/sparkcache/runtime_patches/vllm-kv-block-lease-contract-glm53-b12x-kda-adaptive-mtp.json
@@ -100,7 +100,7 @@
     {
     "path": "vllm/v1/core/single_type_kv_cache_manager.py",
     "accepted_sha256": {
-      "recurrent_boundary_contract": "2ab95dea008d65488bc2d55ccbe023c4481dba633fe92924001b9e4155ff38a2"
+      "recurrent_boundary_contract": "f67a1850a7e0288baaa6d42e7ec55b22b09c156720767e23acaabedcae333c8a"
       },
       "required_symbols": [
         "SingleTypeKVCacheManager.add_local_computed_blocks",

From bd3eec1c10b259a24c5f335161d9f8be51c887cd Mon Sep 17 00:00:00 2001
From: FujitsuPolycom <87842395+FujitsuPolycom@users.noreply.github.com>
Date: Sun, 30 Aug 2026 03:26:14 -0500
Subject: [PATCH 5/6] Ignore proven earlier recurrent checkpoints

vLLM can emit a valid aligned checkpoint while a request is still advancing toward a later SparkCache publication boundary. Treat well-formed entries below the store plan as intermediate evidence: do not latch or poison them. Continue waiting until every recurrent group supplies proof at the exact target boundary. Future, malformed, null, non-recurrent, incomplete, and conflicting target entries remain fail-closed.

Cache namespace impact: none. CacheIdentity values, digest salts, 256-token geometry, manifest schemas, page-delta bytes, and page-tail-cow-v1 are unchanged.

Validation: SparkCache 764 passed, 7 skipped; deploy 108 passed, 1 skipped; Ruff and diff checks passed.
---
 deploy/deepseek_v4/tp4_profile.json         |  2 +-
 deploy/glm52_35bpw/profile.json             |  2 +-
 sparkcache/README.md                        |  8 +++--
 sparkcache/spark_context_cache_connector.py | 38 +++++++++++++--------
 sparkcache/test_defect_regressions.py       | 15 +++++++-
 5 files changed, 44 insertions(+), 21 deletions(-)

diff --git a/deploy/deepseek_v4/tp4_profile.json b/deploy/deepseek_v4/tp4_profile.json
index ce63c32..4fa973c 100644
--- a/deploy/deepseek_v4/tp4_profile.json
+++ b/deploy/deepseek_v4/tp4_profile.json
@@ -4,7 +4,7 @@
   "cache_model_profile": "deepseek-v4-fp8-hma",
   "published_runtime_base": "ghcr.io/fujitsupolycom/gb10-vllm-serving@sha256:6fc26fdad81a18f0fff67ce0a05f6d90165625ea2e1cac8a6f39bfb462017028",
   "sparkcache": {
-    "source_sha256": "155a06101524d4c2d2f55dbbd01576e35d5c729888e216fd2f3963e275949ba0"
+    "source_sha256": "490d2c069c2eb755ecb93727aa47c41df38665427228895af0638b8588a049f3"
   },
   "model": {
     "repository": "deepseek-ai/DeepSeek-V4-Flash-0731",
diff --git a/deploy/glm52_35bpw/profile.json b/deploy/glm52_35bpw/profile.json
index 26876f5..5a972b8 100644
--- a/deploy/glm52_35bpw/profile.json
+++ b/deploy/glm52_35bpw/profile.json
@@ -5,7 +5,7 @@
   "published_runtime_base": "ghcr.io/fujitsupolycom/gb10-vllm-serving@sha256:6fc26fdad81a18f0fff67ce0a05f6d90165625ea2e1cac8a6f39bfb462017028",
   "base_image_requirement": "exact GLM-5.2 3.5-bpw R7 image recorded by the source container inspection",
   "sparkcache": {
-    "source_sha256": "155a06101524d4c2d2f55dbbd01576e35d5c729888e216fd2f3963e275949ba0"
+    "source_sha256": "490d2c069c2eb755ecb93727aa47c41df38665427228895af0638b8588a049f3"
   },
   "model": {
     "repository": "brandonmusic/GLM-5.2-EXL3-TR3v4-3.5bpw-MTP78",
diff --git a/sparkcache/README.md b/sparkcache/README.md
index 3b47425..e763e6f 100644
--- a/sparkcache/README.md
+++ b/sparkcache/README.md
@@ -232,9 +232,11 @@ when placement completes and intentionally excludes that bookkeeping.
   SparkCache defers a new recurrent request until a later cached scheduler step,
   when the preceding forward's hand-off can be observed. It latches one matching
   entry for every recurrent group, including a partial-tail CoW target when the
-  boundary lies inside a recurrent page. Outputs with no entry leave publication
-  pending; incomplete, contradictory, or changed evidence cancels it. A store is
-  emitted only after every recurrent group has a proven pinned block. SparkCache
+  boundary lies inside a recurrent page. Valid entries for an earlier checkpoint
+  are ignored while the request advances; outputs with no target-boundary entry
+  leave publication pending. Incomplete, future, contradictory, or changed
+  target evidence cancels it. A store is emitted only after every recurrent
+  group has a proven pinned block at the exact publication boundary. SparkCache
   never substitutes an accumulated request-table ID because vLLM may have
   replaced that source block while producing the durable CoW target. The
   `sparkcache-page-delta-manifest/v2` schema embeds its authenticated base graph
diff --git a/sparkcache/spark_context_cache_connector.py b/sparkcache/spark_context_cache_connector.py
index 8cf5708..05c4753 100644
--- a/sparkcache/spark_context_cache_connector.py
+++ b/sparkcache/spark_context_cache_connector.py
@@ -1623,14 +1623,16 @@ def _validated_recurrent_boundary_blocks(
     ) -> tuple[tuple[int, int], ...] | None:
         """Validate vLLM's exact recurrent replay-boundary block hand-off.
 
-        vLLM may expose a full-page boundary or a partial-tail CoW target on a
-        later scheduler output than the request which began the store. Absent
-        per-request metadata therefore preserves ``latched`` and keeps the
-        store pending. None means supplied metadata is incomplete,
-        contradictory, or conflicts with an earlier latch, so this publication
-        attempt must be poisoned. SparkCache never derives a replacement from
-        another request-table entry because it can name overwritten running or
-        speculative state instead of vLLM's pinned CoW target.
+        vLLM may expose an earlier aligned checkpoint while the request is
+        still advancing toward this store boundary, followed by a partial-tail
+        CoW target on a later scheduler output. Valid older entries and absent
+        per-request metadata therefore preserve ``latched`` and keep the store
+        pending. None means supplied metadata is incomplete, malformed, ahead
+        of the plan, or conflicts with an earlier same-boundary latch, so this
+        publication attempt must be poisoned. SparkCache never derives a
+        replacement from another request-table entry because it can name
+        overwritten running or speculative state instead of vLLM's pinned CoW
+        target.
         """
 
         def reject(reason: str) -> None:
@@ -1657,9 +1659,11 @@ def reject(reason: str) -> None:
             return latched
         if not isinstance(entries, (list, tuple)):
             return reject("request value is not a sequence")
+        if not entries:
+            return reject("request has no recurrent boundary entries")
 
         overrides: list[tuple[int, int]] = []
-        seen_groups: set[int] = set()
+        seen_target_groups: set[int] = set()
         for entry in entries:
             if not isinstance(entry, (list, tuple)) or len(entry) != 3:
                 return reject("entry is not a group, block, boundary triple")
@@ -1668,18 +1672,22 @@ def reject(reason: str) -> None:
                 return reject("entry values are not integers")
             if not 0 <= group_index < len(self._group_topology):
                 return reject("group index is outside the registered topology")
-            if group_index in seen_groups:
-                return reject("multiple blocks claim the same recurrent group")
             topology = self._group_topology[group_index]
             if topology["reuse_policy"] != "recurrent_align":
                 return reject("group is not an aligned recurrent cache")
             if block_id <= 0:
                 return reject("physical block is vLLM's null block")
-            if entry_boundary != boundary_tokens:
-                return reject("entry boundary differs from the store plan")
-            seen_groups.add(group_index)
+            if entry_boundary < boundary_tokens:
+                continue
+            if entry_boundary > boundary_tokens:
+                return reject("entry boundary is ahead of the store plan")
+            if group_index in seen_target_groups:
+                return reject("multiple blocks claim the same recurrent group")
+            seen_target_groups.add(group_index)
             overrides.append((group_index, block_id))
-        if seen_groups != required_groups:
+        if not overrides:
+            return latched
+        if seen_target_groups != required_groups:
             return reject("entries do not cover every aligned recurrent group")
         validated = tuple(sorted(overrides))
         if latched and validated != latched:
diff --git a/sparkcache/test_defect_regressions.py b/sparkcache/test_defect_regressions.py
index d5a53ae..b46b30f 100644
--- a/sparkcache/test_defect_regressions.py
+++ b/sparkcache/test_defect_regressions.py
@@ -1238,10 +1238,18 @@ def test_nonaligned_boundary_waits_for_partial_tail_cow_mapping(self) -> None:
             output.num_scheduled_tokens[request.req_id] = (
                 self.NONALIGNED_PROMPT_TOKENS
             )
+            output.recurrent_boundary_blocks = {
+                request.req_id: [(1, self.BOUNDARY_BLOCK, self.BOUNDARY)]
+            }
 
             first_metadata = connector.build_connector_meta(output)
             self.assertEqual(first_metadata.plans, [])
             self.assertIn(request.req_id, connector._store_progress)
+            self.assertNotIn(request.req_id, connector._store_recurrent_boundaries)
+            self.assertEqual(
+                connector.counters["recurrent_boundary_metadata_rejected"],
+                0,
+            )
             pending = connector.build_connector_meta(
                 self._cached_scheduler_output(
                     num_computed_tokens=self.NONALIGNED_PROMPT_TOKENS,
@@ -1337,6 +1345,11 @@ def test_chunked_prefill_validates_only_at_publication_step(self) -> None:
             middle = self._cached_scheduler_output(
                 num_computed_tokens=2304,
                 num_scheduled_tokens=2304,
+                recurrent_boundary_blocks={
+                    "dflash-recurrent-boundary": [
+                        (1, self.BOUNDARY_BLOCK - 1, 2304)
+                    ]
+                },
             )
             self.assertEqual(connector.build_connector_meta(middle).plans, [])
             self.assertIn("dflash-recurrent-boundary", connector._store_progress)
@@ -1423,7 +1436,7 @@ def test_missing_or_wrong_request_metadata_keeps_publication_pending(self) -> No
     def test_contradictory_boundary_metadata_skips_publication(self) -> None:
         invalid_entries = (
             [],
-            [(1, self.BOUNDARY_BLOCK, self.BOUNDARY - 256)],
+            [(1, self.BOUNDARY_BLOCK, self.BOUNDARY + 256)],
             [(2, self.BOUNDARY_BLOCK, self.BOUNDARY)],
             [(1, self.BOUNDARY_BLOCK, self.BOUNDARY), (1, 43, self.BOUNDARY)],
             [(1, 0, self.BOUNDARY)],

From c56f77f97b3da907d32e888d82046359a62f0f88 Mon Sep 17 00:00:00 2001
From: FujitsuPolycom <87842395+FujitsuPolycom@users.noreply.github.com>
Date: Sun, 30 Aug 2026 03:50:28 -0500
Subject: [PATCH 6/6] Log future recurrent proof identity

Keep future-boundary evidence fail-closed, but include its observed boundary, target boundary, recurrent group, and physical block in the rejection reason. This makes the live scheduler hand-off diagnosable without changing acceptance semantics.

Cache namespace impact: none. CacheIdentity values, digest salts, chunk geometry, manifests, page deltas, and page-tail-cow-v1 are unchanged.

Validation: SparkCache 765 passed, 7 skipped; deploy 108 passed, 1 skipped; Ruff and diff checks passed.
---
 deploy/deepseek_v4/tp4_profile.json         |  2 +-
 deploy/glm52_35bpw/profile.json             |  2 +-
 sparkcache/spark_context_cache_connector.py |  6 ++-
 sparkcache/test_defect_regressions.py       | 46 +++++++++++++++++++++
 4 files changed, 53 insertions(+), 3 deletions(-)

diff --git a/deploy/deepseek_v4/tp4_profile.json b/deploy/deepseek_v4/tp4_profile.json
index 4fa973c..cda756a 100644
--- a/deploy/deepseek_v4/tp4_profile.json
+++ b/deploy/deepseek_v4/tp4_profile.json
@@ -4,7 +4,7 @@
   "cache_model_profile": "deepseek-v4-fp8-hma",
   "published_runtime_base": "ghcr.io/fujitsupolycom/gb10-vllm-serving@sha256:6fc26fdad81a18f0fff67ce0a05f6d90165625ea2e1cac8a6f39bfb462017028",
   "sparkcache": {
-    "source_sha256": "490d2c069c2eb755ecb93727aa47c41df38665427228895af0638b8588a049f3"
+    "source_sha256": "788686e858ba4af01f535e95122c7650f412fddc40cd221a0924f4ce2b32ff98"
   },
   "model": {
     "repository": "deepseek-ai/DeepSeek-V4-Flash-0731",
diff --git a/deploy/glm52_35bpw/profile.json b/deploy/glm52_35bpw/profile.json
index 5a972b8..d8e66c8 100644
--- a/deploy/glm52_35bpw/profile.json
+++ b/deploy/glm52_35bpw/profile.json
@@ -5,7 +5,7 @@
   "published_runtime_base": "ghcr.io/fujitsupolycom/gb10-vllm-serving@sha256:6fc26fdad81a18f0fff67ce0a05f6d90165625ea2e1cac8a6f39bfb462017028",
   "base_image_requirement": "exact GLM-5.2 3.5-bpw R7 image recorded by the source container inspection",
   "sparkcache": {
-    "source_sha256": "490d2c069c2eb755ecb93727aa47c41df38665427228895af0638b8588a049f3"
+    "source_sha256": "788686e858ba4af01f535e95122c7650f412fddc40cd221a0924f4ce2b32ff98"
   },
   "model": {
     "repository": "brandonmusic/GLM-5.2-EXL3-TR3v4-3.5bpw-MTP78",
diff --git a/sparkcache/spark_context_cache_connector.py b/sparkcache/spark_context_cache_connector.py
index 05c4753..4803044 100644
--- a/sparkcache/spark_context_cache_connector.py
+++ b/sparkcache/spark_context_cache_connector.py
@@ -1680,7 +1680,11 @@ def reject(reason: str) -> None:
             if entry_boundary < boundary_tokens:
                 continue
             if entry_boundary > boundary_tokens:
-                return reject("entry boundary is ahead of the store plan")
+                return reject(
+                    "entry boundary is ahead of the store plan"
+                    f" observed={entry_boundary} target={boundary_tokens}"
+                    f" group={group_index} block={block_id}"
+                )
             if group_index in seen_target_groups:
                 return reject("multiple blocks claim the same recurrent group")
             seen_target_groups.add(group_index)
diff --git a/sparkcache/test_defect_regressions.py b/sparkcache/test_defect_regressions.py
index b46b30f..c7d3a1a 100644
--- a/sparkcache/test_defect_regressions.py
+++ b/sparkcache/test_defect_regressions.py
@@ -1478,6 +1478,52 @@ def test_contradictory_boundary_metadata_skips_publication(self) -> None:
                         1,
                     )
 
+    def test_future_boundary_rejection_logs_observed_mapping_identity(self) -> None:
+        output = self._scheduler_output()
+        request = output.scheduled_new_reqs[0]
+        request.prompt_token_ids = list(range(self.NONALIGNED_PROMPT_TOKENS))
+        output.num_scheduled_tokens[request.req_id] = self.NONALIGNED_PROMPT_TOKENS
+        with tempfile.TemporaryDirectory() as directory:
+            connector = _make_connector(
+                Path(directory),
+                0,
+                block_size=256,
+                role=KVConnectorRole.SCHEDULER,
+                override_worker_rank=False,
+                tp=1,
+                dcp=1,
+                kv_cache_config=self._config(),
+                extra_config={"spark_cache_model_profile": "glm53-flash-hybrid"},
+            )
+            connector.build_connector_meta(output)
+
+            with mock.patch.object(
+                connector_module.logger, "warning"
+            ) as warning:
+                metadata = connector.build_connector_meta(
+                    self._cached_scheduler_output(
+                        num_computed_tokens=self.NONALIGNED_PROMPT_TOKENS,
+                        recurrent_boundary_blocks={
+                            request.req_id: [
+                                (
+                                    1,
+                                    self.COW_BLOCK,
+                                    self.NONALIGNED_BOUNDARY + 256,
+                                )
+                            ]
+                        },
+                    )
+                )
+
+            self.assertEqual(metadata.plans, [])
+            warning.assert_called_once()
+            log_args = warning.call_args.args
+            self.assertIn(
+                "entry boundary is ahead of the store plan"
+                " observed=8448 target=8192 group=1 block=142",
+                log_args[0] % log_args[1:],
+            )
+
     def test_partial_recurrent_group_coverage_skips_publication(self) -> None:
         config = self._config()
         second_recurrent = types.SimpleNamespace(