From de021e8eff2bdaeb4dda8bd26c608c7f37eea3eb Mon Sep 17 00:00:00 2001
From: FujitsuPolycom <87842395+FujitsuPolycom@users.noreply.github.com>
Date: Sun, 6 Sep 2026 17:44:22 -0500
Subject: [PATCH 1/7] Attribute accepted prompt work from scheduler lifecycle
 events

Count consumed local prefixes, verified external reuse, and accepted target prompt computation across preemption attempts. Optional scheduler callbacks prevent offers, cancelled restores, and late worker drains from becoming reuse credit. Emit bounded per-request diagnostics only when reuse tracing is enabled. Cache identity and namespace are unchanged. Validation: 1,136 CPU tests passed with eight documented skips, 22 focused attribution tests passed, and Ruff passed.
---
 deploy/deepseek_v4/tp4_profile.json         |   4 +-
 deploy/glm52_35bpw/profile.json             |   2 +-
 sparkcache/README.md                        |  33 +++++
 sparkcache/request_attribution.py           | 122 +++++++++++++++++
 sparkcache/spark_context_cache_connector.py |  52 ++++++++
 sparkcache/test_request_attribution.py      | 141 ++++++++++++++++++++
 sparkcache/test_reuse_trace.py              |  56 ++++++++
 7 files changed, 407 insertions(+), 3 deletions(-)
 create mode 100644 sparkcache/request_attribution.py
 create mode 100644 sparkcache/test_request_attribution.py

diff --git a/deploy/deepseek_v4/tp4_profile.json b/deploy/deepseek_v4/tp4_profile.json
index a6a7f07..d89b1d2 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": "72f33311d1ee5811c2b3e4eefda98c9d17b5afe8d9d015a2e3c26c893f25aabc"
+    "source_sha256": "5c0395839d4a1b4bcc9955482830c6050127886c14ec2e6663242ba251e6f090"
   },
   "model": {
     "repository": "deepseek-ai/DeepSeek-V4-Flash-0731",
@@ -23,7 +23,7 @@
     "kv_cache_bytes_per_rank": 34359738368,
     "max_model_len": 524288,
     "max_num_seqs": 32,
-    "gpu_memory_utilization": 0.70,
+    "gpu_memory_utilization": 0.7,
     "tokenizer_mode": "deepseek_v4",
     "speculation_method": "dspark",
     "speculation_tokens": 5,
diff --git a/deploy/glm52_35bpw/profile.json b/deploy/glm52_35bpw/profile.json
index 0bbbfb9..88dd665 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": "72f33311d1ee5811c2b3e4eefda98c9d17b5afe8d9d015a2e3c26c893f25aabc"
+    "source_sha256": "5c0395839d4a1b4bcc9955482830c6050127886c14ec2e6663242ba251e6f090"
   },
   "model": {
     "repository": "brandonmusic/GLM-5.2-EXL3-TR3v4-3.5bpw-MTP78",
diff --git a/sparkcache/README.md b/sparkcache/README.md
index 768f7de..6593521 100644
--- a/sparkcache/README.md
+++ b/sparkcache/README.md
@@ -424,6 +424,39 @@ The counters describe host-side operations. They do not report filesystem
 allocation, NVMe Data Units Written, controller write amplification, or NAND
 writes.
 
+### Request reuse attribution
+
+Status: **implemented** with an optional scheduler callback. A runtime without
+that callback cannot produce exact request attribution from connector offers.
+
+Set `SPARK_CONTEXT_CACHE_TRACE_REUSE=1` before startup. An instrumented scheduler
+emits one `request_cache_attribution` event in `sparkcache-reuse-trace/v1` at
+request cleanup. No request IDs are added to Prometheus labels.
+
+| Field | Meaning |
+|---|---|
+| `local_tokens_reused` | GPU-resident prompt tokens consumed by accepted target execution, including resident shared leases. |
+| `external_tokens_reused` | Prompt tokens consumed after a successful all-rank persistent restore and the scheduler's final-token adjustment. |
+| `prompt_tokens_computed` | Prompt intervals completed by accepted target execution, accumulated across preemption attempts. |
+| `preemptions` | Request preemption generation observed by the scheduler. |
+| `attribution_complete` | True only for normal completion with observed prompt completion and no missing or invalid accounting boundary. |
+
+Counts cover accepted target-prompt work across attempts. They can exceed the
+original prompt length after preemption. They exclude output tokens, draft
+execution, replay inside kernels, and rejected worker output.
+
+An offered restore earns no credit. A verified restore aborted before target
+execution earns no reused-token credit. A follower consuming a resident GPU
+lease records local reuse, even if a different request restored that lease.
+
+The restored state span and external prompt tokens reused are distinct. A
+restore can write an already-local prefix, and a full prompt hit still needs
+the final prompt token recomputed for sampling logits.
+
+Incomplete observations remain labeled incomplete. Their token fields are not
+added to the connector's `attribution_completed_*` aggregate counters. Request
+cleanup releases the ledger even when logging fails.
+
 Telemetry is observational. It cannot change publication, restore, cache
 identity, or serving decisions.
 
diff --git a/sparkcache/request_attribution.py b/sparkcache/request_attribution.py
new file mode 100644
index 0000000..3a9c2f8
--- /dev/null
+++ b/sparkcache/request_attribution.py
@@ -0,0 +1,122 @@
+"""Attribute accepted target-prompt work from authoritative scheduler events.
+
+Reuse is credited when accepted execution consumes a selected prefix. Counts
+accumulate across preemption attempts and can exceed the original prompt size.
+They exclude draft work, replay inside kernels, and rejected worker outputs.
+"""
+
+from dataclasses import dataclass
+
+
+def _count(value: object) -> int:
+    if type(value) is not int or value < 0:
+        raise ValueError("Attribution token counts and generations must be nonnegative integers")
+    return value
+
+
+@dataclass
+class _Attempt:
+    generation: int
+    local: int
+    offered_external: int
+    external: int = 0
+    pending_restore: bool = False
+    consumed: bool = False
+    progress: int = 0
+    can_readmit: bool = False
+
+
+class RequestAttribution:
+    def __init__(self, prompt_tokens: int):
+        self.prompt_tokens = _count(prompt_tokens)
+        self.local_tokens_reused = 0
+        self.external_tokens_reused = 0
+        self.prompt_tokens_computed = 0
+        self.preemptions = 0
+        self.valid = True
+        self.prompt_completed = False
+        self.attempt: _Attempt | None = None
+
+    def record(self, event: str, **fields) -> None:
+        generation = _count(fields.get("preemptions", 0))
+        if generation < self.preemptions:
+            self.valid = False
+            return
+        if event == "preempted":
+            self.preemptions = generation
+            self.attempt = None
+            return
+        if event == "admitted":
+            local = min(_count(fields["local_tokens"]), self.prompt_tokens)
+            external = min(_count(fields["external_tokens"]), self.prompt_tokens - local)
+            if (self.attempt is not None and generation == self.attempt.generation
+                    and not self.attempt.can_readmit):
+                # Resuming a parked restore reports no newly allocated prefix.
+                # Its original allocation and receive outcome remain authoritative.
+                return
+            if generation != self.preemptions:
+                self.valid = False
+            self.preemptions = generation
+            self.attempt = _Attempt(generation, local, external,
+                                    pending_restore=external > 0)
+            return
+        attempt = self.attempt
+        if attempt is None or generation != attempt.generation:
+            self.valid = False
+            return
+        if event == "restore_finalized":
+            prefix = min(_count(fields["valid_prefix_tokens"]), self.prompt_tokens)
+            if not attempt.pending_restore or attempt.consumed:
+                self.valid = False
+                return
+            attempt.pending_restore = False
+            attempt.can_readmit = fields.get("success") is False
+            attempt.local = min(attempt.local, prefix)
+            accepted = max(0, prefix - attempt.local)
+            if accepted > attempt.offered_external:
+                self.valid = False
+            if fields.get("success") is True:
+                attempt.external = min(accepted, attempt.offered_external)
+            elif accepted:
+                # A failed restore cannot prove that its residual external
+                # prefix passed every rank's integrity boundary.
+                self.valid = False
+            return
+        if event != "prompt_step_completed":
+            raise ValueError("Unknown request attribution event")
+        start = min(_count(fields["start_token"]), self.prompt_tokens)
+        end = min(_count(fields["end_token"]), self.prompt_tokens)
+        if end < start:
+            raise ValueError("Completed prompt interval ends before it starts")
+        if fields.get("stale"):
+            return
+        if attempt.pending_restore:
+            self.valid = False
+        prefix = attempt.local + attempt.external
+        if start > max(prefix, attempt.progress):
+            self.valid = False
+        if not attempt.consumed:
+            # Work that recomputes an admitted prefix must not also claim that
+            # same prefix as consumed reuse.
+            used_local = min(attempt.local, start)
+            used_external = min(attempt.external, max(0, start - used_local))
+            self.local_tokens_reused += used_local
+            self.external_tokens_reused += used_external
+            attempt.consumed = True
+        self.prompt_tokens_computed += max(0, end - max(start, attempt.progress))
+        attempt.progress = max(attempt.progress, end)
+        self.prompt_completed |= end == self.prompt_tokens
+
+    def summary(self, status: str) -> dict:
+        return {
+            "prompt_tokens": self.prompt_tokens,
+            "local_tokens_reused": self.local_tokens_reused,
+            "external_tokens_reused": self.external_tokens_reused,
+            "prompt_tokens_computed": self.prompt_tokens_computed,
+            "preemptions": self.preemptions,
+            "attribution_complete": self.valid and self.prompt_completed and status in (
+                "FINISHED_STOPPED", "FINISHED_LENGTH_CAPPED",
+            ),
+            "status": status,
+            "accounting_scope": "accepted_target_prompt_work_across_attempts",
+        }
diff --git a/sparkcache/spark_context_cache_connector.py b/sparkcache/spark_context_cache_connector.py
index ec00a7f..b390718 100644
--- a/sparkcache/spark_context_cache_connector.py
+++ b/sparkcache/spark_context_cache_connector.py
@@ -94,6 +94,7 @@
     split_snapshot,
 )
 from sparkcache.spark_context_cache_restore_timing import RestoreTiming
+from sparkcache.request_attribution import RequestAttribution
 from sparkcache.held_inventory import HeldInventory
 from sparkcache.page_base_read_flights import (
     PageBaseReadEvidence,
@@ -961,6 +962,10 @@ def __init__(
         )
         self._config = config
         self._trace_reuse_enabled = os.environ.get("SPARK_CONTEXT_CACHE_TRACE_REUSE") == "1"
+        self.request_cache_events_enabled = (
+            self._trace_reuse_enabled and role is KVConnectorRole.SCHEDULER
+        )
+        self._request_attribution: dict[str, RequestAttribution] = {}
         self._block_size = config.block_size
         self._tp_degree = config.tp_degree
         self._dcp_degree = config.dcp_degree
@@ -1977,6 +1982,52 @@ def _trace_reuse(
                 json.dumps(record, sort_keys=True, separators=(",", ":")),
             )
 
+    def record_request_cache_event(self, request: "Request", event: str, **fields) -> None:
+        """Consume optional scheduler observations without changing serving.
+
+        Matching offers and worker-local completion traces never call this
+        accounting path. The runtime must report adopted prefixes, final
+        receive outcomes, and accepted target-model execution explicitly.
+        """
+        if not getattr(self, "request_cache_events_enabled", False):
+            return
+        request_id = request.request_id
+        state = self._request_attribution.get(request_id)
+        try:
+            if event == "finished":
+                state = self._request_attribution.pop(request_id, None)
+                if state is None:
+                    return
+                status = fields.get("status", getattr(request, "status", "unknown"))
+                status = str(getattr(status, "name", status))
+                summary = state.summary(status)
+                category = "complete" if summary["attribution_complete"] else "incomplete"
+                key = f"attribution_requests_{category}"
+                self.counters[key] = self.counters.get(key, 0) + 1
+                if summary["attribution_complete"]:
+                    for field in ("local_tokens_reused", "external_tokens_reused",
+                                  "prompt_tokens_computed"):
+                        key = "attribution_completed_" + field
+                        self.counters[key] = self.counters.get(key, 0) + summary[field]
+                self._trace_reuse("request_cache_attribution", request_id, **summary)
+                return
+            if state is None:
+                if event != "admitted":
+                    # Late receive drains for retired requests do not allocate
+                    # another ledger or credit an aborted response.
+                    return
+                prompt_tokens = getattr(request, "num_prompt_tokens", None)
+                if prompt_tokens is None:
+                    prompt_tokens = len(request.prompt_token_ids or ())
+                state = RequestAttribution(prompt_tokens)
+                self._request_attribution[request_id] = state
+            state.record(event, **fields)
+        except Exception:  # Diagnostics cannot interrupt verified-or-recompute.
+            if state is not None:
+                state.valid = False
+            key = "attribution_invalid_events"
+            self.counters[key] = self.counters.get(key, 0) + 1
+
     def shared_prefix_lease_attached(self, request_id: str, lease_key: str) -> None:
         follower = self._restore_flight_followers.get(request_id)
         if follower is not None and follower.lease_digest == lease_key:
@@ -4940,6 +4991,7 @@ def request_finished(
         request: "Request",
         block_ids: list[int],
     ) -> tuple[bool, dict[str, Any] | None]:
+        self.record_request_cache_event(request, "finished")
         # A finished request id never recurs, so its scheduler-side tracking
         # state is dropped here. This is what keeps _need_load, _admitted,
         # and _store_progress bounded without evicting live entries.
diff --git a/sparkcache/test_request_attribution.py b/sparkcache/test_request_attribution.py
new file mode 100644
index 0000000..b020045
--- /dev/null
+++ b/sparkcache/test_request_attribution.py
@@ -0,0 +1,141 @@
+"""Prompt attribution follows accepted execution, not matching offers."""
+
+import pytest
+
+from sparkcache.request_attribution import RequestAttribution
+
+
+def admit(state, local=0, external=0, generation=0):
+    state.record("admitted", local_tokens=local, external_tokens=external,
+                 preemptions=generation, source="prefix_lookup")
+
+
+def completed(state, start, end, generation=0):
+    state.record("prompt_step_completed", start_token=start, end_token=end,
+                 preemptions=generation)
+
+
+def test_local_partial_tail_uses_final_scheduler_count():
+    state = RequestAttribution(1100)
+    admit(state, local=1031)
+    completed(state, 1031, 1100)
+    assert state.summary("FINISHED_STOPPED") == {
+        "prompt_tokens": 1100, "local_tokens_reused": 1031,
+        "external_tokens_reused": 0, "prompt_tokens_computed": 69,
+        "preemptions": 0, "attribution_complete": True,
+        "status": "FINISHED_STOPPED",
+        "accounting_scope": "accepted_target_prompt_work_across_attempts",
+    }
+
+
+def test_external_offer_without_finalization_is_not_reuse():
+    state = RequestAttribution(1100)
+    admit(state, local=256, external=768)
+    completed(state, 256, 1100)
+    result = state.summary("FINISHED_STOPPED")
+    assert result["external_tokens_reused"] == 0
+    assert not result["attribution_complete"]
+
+
+def test_verified_restore_counts_only_effective_prompt_reuse_once():
+    state = RequestAttribution(1024)
+    admit(state, local=256, external=768)
+    state.record("restore_finalized", valid_prefix_tokens=1023,
+                 success=True, preemptions=0)
+    # Re-admission of the parked request does not replace its source accounting.
+    admit(state)
+    completed(state, 1023, 1024)
+    completed(state, 1023, 1024)
+    result = state.summary("FINISHED_STOPPED")
+    assert (result["local_tokens_reused"], result["external_tokens_reused"],
+            result["prompt_tokens_computed"]) == (256, 767, 1)
+    assert result["attribution_complete"]
+
+
+def test_failed_restore_recomputes_and_does_not_credit_external_offer():
+    state = RequestAttribution(1100)
+    admit(state, local=256, external=768)
+    state.record("restore_finalized", valid_prefix_tokens=256,
+                 success=False, preemptions=0)
+    completed(state, 256, 1100)
+    result = state.summary("FINISHED_STOPPED")
+    assert result["external_tokens_reused"] == 0
+    assert result["prompt_tokens_computed"] == 844
+    assert result["attribution_complete"]
+
+
+def test_gpu_lease_is_local_reuse_without_external_credit():
+    state = RequestAttribution(1024)
+    state.record("admitted", local_tokens=1023, external_tokens=0,
+                 preemptions=0, source="gpu_lease")
+    completed(state, 1023, 1024)
+    result = state.summary("FINISHED_STOPPED")
+    assert result["local_tokens_reused"] == 1023
+    assert result["external_tokens_reused"] == 0
+
+
+def test_failed_private_restore_allows_fresh_local_lookup_in_same_generation():
+    state = RequestAttribution(1100)
+    admit(state, external=1024)
+    state.record("restore_finalized", valid_prefix_tokens=0,
+                 success=False, preemptions=0)
+    admit(state, local=768)
+    completed(state, 768, 1100)
+    result = state.summary("FINISHED_STOPPED")
+    assert result["external_tokens_reused"] == 0
+    assert result["local_tokens_reused"] == 768
+    assert result["prompt_tokens_computed"] == 332
+    assert result["attribution_complete"]
+
+
+def test_preemption_keeps_completed_work_and_counts_recomputation():
+    state = RequestAttribution(1000)
+    admit(state, local=256)
+    completed(state, 256, 512)
+    state.record("preempted", preemptions=1)
+    admit(state, generation=1)
+    completed(state, 0, 1000, generation=1)
+    result = state.summary("FINISHED_STOPPED")
+    assert result["local_tokens_reused"] == 256
+    assert result["prompt_tokens_computed"] == 1256
+    assert result["preemptions"] == 1
+    assert result["attribution_complete"]
+
+
+def test_abort_before_execution_does_not_credit_verified_cache_warming():
+    state = RequestAttribution(1024)
+    admit(state, external=1024)
+    state.record("restore_finalized", valid_prefix_tokens=1023,
+                 success=True, preemptions=0)
+    result = state.summary("FINISHED_ABORTED")
+    assert result["external_tokens_reused"] == 0
+    assert result["prompt_tokens_computed"] == 0
+    assert not result["attribution_complete"]
+
+
+def test_decode_after_preemption_credits_resident_prompt_without_output_tokens():
+    state = RequestAttribution(1000)
+    admit(state)
+    completed(state, 0, 1000)
+    state.record("preempted", preemptions=1)
+    admit(state, local=1000, generation=1)
+    completed(state, 1000, 1000, generation=1)
+    result = state.summary("FINISHED_STOPPED")
+    assert result["local_tokens_reused"] == 1000
+    assert result["prompt_tokens_computed"] == 1000
+
+
+def test_missing_attempt_and_unobserved_prefix_do_not_claim_complete_attribution():
+    state = RequestAttribution(1000)
+    completed(state, 200, 1000)
+    assert not state.summary("FINISHED_STOPPED")["attribution_complete"]
+
+
+@pytest.mark.parametrize("field,value", [("local_tokens", -1),
+                                          ("external_tokens", True),
+                                          ("preemptions", -1)])
+def test_invalid_scheduler_observation_is_rejected(field, value):
+    fields = dict(local_tokens=0, external_tokens=0, preemptions=0)
+    fields[field] = value
+    with pytest.raises(ValueError):
+        RequestAttribution(1000).record("admitted", **fields)
diff --git a/sparkcache/test_reuse_trace.py b/sparkcache/test_reuse_trace.py
index e143df5..d3a16e5 100644
--- a/sparkcache/test_reuse_trace.py
+++ b/sparkcache/test_reuse_trace.py
@@ -122,3 +122,59 @@ def fail(*args, **kwargs):
         assert connector.counters["shared_prefix_leases_attached"] == 1
     finally:
         connector.shutdown()
+
+
+def test_request_attribution_credits_scheduler_events_and_cleans_up(tmp_path, monkeypatch):
+    monkeypatch.setenv("SPARK_CONTEXT_CACHE_TRACE_REUSE", "1")
+    records = _records(monkeypatch)
+    connector = fixtures._make_connector(tmp_path, 0, role=connector_module.KVConnectorRole.SCHEDULER)
+    request = SimpleNamespace(request_id="attributed", num_prompt_tokens=1100,
+                              status=SimpleNamespace(name="FINISHED_STOPPED"))
+    try:
+        assert connector.request_cache_events_enabled
+        connector.record_request_cache_event(request, "admitted", local_tokens=1031,
+                                             external_tokens=0, preemptions=0)
+        connector.record_request_cache_event(request, "prompt_step_completed",
+                                             start_token=1031, end_token=1100, preemptions=0)
+        connector.record_request_cache_event(request, "finished")
+        connector.record_request_cache_event(request, "finished")
+        connector.record_request_cache_event(request, "restore_finalized",
+                                             success=True, valid_prefix_tokens=1024)
+        assert connector._request_attribution == {}
+        assert len(records) == 1
+        assert records[0]["local_tokens_reused"] == 1031
+        assert records[0]["prompt_tokens_computed"] == 69
+        assert records[0]["external_tokens_reused"] == 0
+        assert records[0]["attribution_complete"]
+        assert connector.counters["attribution_requests_complete"] == 1
+    finally:
+        connector.shutdown()
+
+
+def test_request_attribution_invalid_event_and_log_failure_do_not_affect_serving(tmp_path, monkeypatch):
+    monkeypatch.setenv("SPARK_CONTEXT_CACHE_TRACE_REUSE", "1")
+    connector = fixtures._make_connector(tmp_path, 0, role=connector_module.KVConnectorRole.SCHEDULER)
+    request = SimpleNamespace(request_id="invalid", num_prompt_tokens=100,
+                              status=SimpleNamespace(name="FINISHED_ABORTED"))
+    try:
+        connector.record_request_cache_event(request, "admitted", local_tokens=-1,
+                                             external_tokens=0, preemptions=0)
+        assert connector.counters["attribution_invalid_events"] == 1
+        monkeypatch.setattr(connector_module.logger, "info",
+                            lambda *args, **kwargs: (_ for _ in ()).throw(OSError("sink failed")))
+        connector.record_request_cache_event(request, "finished")
+        assert connector._request_attribution == {}
+        assert connector.counters["attribution_requests_incomplete"] == 1
+    finally:
+        connector.shutdown()
+
+
+def test_request_attribution_disabled_has_no_per_request_state(tmp_path, monkeypatch):
+    monkeypatch.delenv("SPARK_CONTEXT_CACHE_TRACE_REUSE", raising=False)
+    connector = fixtures._make_connector(tmp_path, 0, role=connector_module.KVConnectorRole.SCHEDULER)
+    try:
+        assert not connector.request_cache_events_enabled
+        connector.record_request_cache_event(SimpleNamespace(request_id="off"), "admitted")
+        assert connector._request_attribution == {}
+    finally:
+        connector.shutdown()

From 19873f697c1ebdaf2b11d2013f2411b31f9e0f81 Mon Sep 17 00:00:00 2001
From: FujitsuPolycom <87842395+FujitsuPolycom@users.noreply.github.com>
Date: Sun, 6 Sep 2026 17:42:15 -0500
Subject: [PATCH 2/7] Pace cache deletion work with per-pass limits and
 cooldown

Add opt-in unlink-attempt limits and cooldown between maintenance passes. Preserve protected reference graphs and durable root barriers; reclaim orphan debt before selecting further live roots. Expose deferred-work and cooldown telemetry. Full inventory scans remain unbounded; defaults, cache identities, and persisted formats are unchanged.

Validation: 1136 SparkCache tests passed with 8 skips; 111 deployment tests passed with 1 skip; Ruff clean. GPU and serving performance qualification remains pending.
---
 deploy/deepseek_v4/tp4_profile.json           |   2 +-
 deploy/glm52_35bpw/profile.json               |   2 +-
 sparkcache/README.md                          |  57 ++++++
 .../cache_manifest.py                         |  57 +++++-
 .../test_maintenance_budget.py                | 175 ++++++++++++++++++
 sparkcache/spark_context_cache_config.py      |  10 +
 sparkcache/spark_context_cache_connector.py   |  49 ++++-
 sparkcache/test_maintenance_pacing.py         |  61 ++++++
 sparkcache/test_spark_context_cache_config.py |   4 +
 9 files changed, 408 insertions(+), 9 deletions(-)
 create mode 100644 sparkcache/persistent_context_cache/test_maintenance_budget.py
 create mode 100644 sparkcache/test_maintenance_pacing.py

diff --git a/deploy/deepseek_v4/tp4_profile.json b/deploy/deepseek_v4/tp4_profile.json
index d89b1d2..7e25529 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": "5c0395839d4a1b4bcc9955482830c6050127886c14ec2e6663242ba251e6f090"
+    "source_sha256": "6ecc20725a23c4d260cf227b1d2805fe925d3def471caa0ad7b33267f704da76"
   },
   "model": {
     "repository": "deepseek-ai/DeepSeek-V4-Flash-0731",
diff --git a/deploy/glm52_35bpw/profile.json b/deploy/glm52_35bpw/profile.json
index 88dd665..d71763a 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": "5c0395839d4a1b4bcc9955482830c6050127886c14ec2e6663242ba251e6f090"
+    "source_sha256": "6ecc20725a23c4d260cf227b1d2805fe925d3def471caa0ad7b33267f704da76"
   },
   "model": {
     "repository": "brandonmusic/GLM-5.2-EXL3-TR3v4-3.5bpw-MTP78",
diff --git a/sparkcache/README.md b/sparkcache/README.md
index 6593521..69fb344 100644
--- a/sparkcache/README.md
+++ b/sparkcache/README.md
@@ -311,6 +311,63 @@ A larger gap amortizes maintenance across more writes, but each pass evicts
 more data and can increase future misses. Compare those costs with observed
 publication age and maintenance activity before changing the gap.
 
+### Deletion pacing
+
+Status: **implemented**; serving performance with these controls is
+**research-only**. Both settings default to `0`, preserving unrestricted passes:
+
+- `spark_cache_maintenance_max_deletions` caps filesystem unlink attempts per
+  pass, including failed attempts, across manifests, aliases, debris,
+  descriptor segments, and payload objects. Its environment fallback is
+  `SPARK_CONTEXT_CACHE_MAINTENANCE_MAX_DELETIONS`.
+- `spark_cache_maintenance_interval_ms` sets a minimum cooldown after a pass
+  finishes or fails. Forced post-commit calls also respect it; skipped calls
+  do not extend it. Its environment fallback is
+  `SPARK_CONTEXT_CACHE_MAINTENANCE_INTERVAL_MS`.
+
+Explicit connector settings take precedence over environment values. A test
+configuration can select `128` deletion attempts and `1000` milliseconds;
+these values are not a qualified serving-performance recommendation.
+
+With a deletion budget, existing orphan payloads and debris are reclaimed
+before additional roots are selected.
+
+A pass can stop above the low watermark
+or the capacity maximum; optional store admission remains blocked while
+capacity is unsatisfied.
+
+Background retries complete deferred cleanup without
+making serving wait. Root-directory durability barriers still precede object
+removal, and protected publication roots retain their complete object graphs.
+
+This is **not a scan-size or wall-clock bound**. Each admitted pass authenticates
+the complete reference inventory and reconciles survivors.
+
+One filesystem
+operation or durability barrier can take arbitrarily long. Smaller deletion
+budgets can increase total scan work.
+
+Cooldown trades reclamation throughput
+for gaps between that work. A larger watermark gap does not remove this cost.
+
+`MaintenanceReport.deletion_attempts` counts admitted unlink attempts;
+`work_pending` reports cleanup deferred by the budget or orphan-first policy;
+`skipped_cooldown` reports a pass skipped before inventory work.
+
+Connector
+counters `capacity_deletion_attempts`, `capacity_budget_exhausted`, and
+`capacity_skipped_cooldown` expose the same activity. Capacity log records
+include `deletion_attempts` and `work_pending`.
+
+Prometheus gauges
+`vllm:sparkcache_maintenance_deletion_attempts`,
+`vllm:sparkcache_maintenance_budget_exhausted`, and
+`vllm:sparkcache_maintenance_skipped_cooldown` sum reported cumulative counts
+across ranks.
+
+They reset with workers and reflect the last worker reports,
+not independent scrape-time measurements.
+
 `spark_cache_ttl_seconds` expires manifests by recency; zero disables TTL.
 Maintenance preserves shared objects referenced by surviving manifests.
 
diff --git a/sparkcache/persistent_context_cache/cache_manifest.py b/sparkcache/persistent_context_cache/cache_manifest.py
index dc82631..45a87ca 100644
--- a/sparkcache/persistent_context_cache/cache_manifest.py
+++ b/sparkcache/persistent_context_cache/cache_manifest.py
@@ -444,9 +444,12 @@ class CapacityPolicy:
     max_bytes: int = 0
     low_watermark_bytes: int = 0
     ttl_seconds: int = 0
+    maintenance_max_deletions: int = 0
+    maintenance_interval_ms: int = 0
 
     def __post_init__(self) -> None:
-        for field in ("max_bytes", "low_watermark_bytes", "ttl_seconds"):
+        for field in ("max_bytes", "low_watermark_bytes", "ttl_seconds",
+                      "maintenance_max_deletions", "maintenance_interval_ms"):
             value = getattr(self, field)
             if isinstance(value, bool) or not isinstance(value, int) or value < 0:
                 raise ValueError(f"{field} must be a non-negative integer")
@@ -473,6 +476,9 @@ class MaintenanceReport:
     evicted_entries: tuple[EntryKey, ...] = ()
     capacity_satisfied: bool = True
     skipped_busy: bool = False
+    skipped_cooldown: bool = False
+    deletion_attempts: int = 0
+    work_pending: bool = False
     aliases_evicted: int = 0
     segments_deleted: int = 0
     orphan_segments_deleted: int = 0
@@ -2410,7 +2416,22 @@ def maintain(
             guard.__enter__()
         except BlockingIOError:
             return MaintenanceReport(capacity_satisfied=False, skipped_busy=True)
+        if time.monotonic() < getattr(self, "_maintenance_not_before", 0.0):
+            guard.__exit__(None, None, None)
+            return MaintenanceReport(capacity_satisfied=False, skipped_cooldown=True)
         try:
+            deletion_attempts = 0
+            work_pending = False
+
+            def admit_deletion() -> bool:
+                nonlocal deletion_attempts, work_pending
+                if (policy.maintenance_max_deletions > 0
+                        and deletion_attempts >= policy.maintenance_max_deletions):
+                    work_pending = True
+                    return False
+                deletion_attempts += 1
+                return True
+
             manifests_root = self.root / "manifests"
             aliases_root = self.root / "prefix-aliases"
             manifest_paths = (
@@ -2526,17 +2547,36 @@ def root_files(root: Path) -> tuple[Path, ...]:
                     or references.get(path.stem, 0) == 0
                 )
             )
+            # With a deletion budget, reclaim already-unreferenced payloads
+            # before choosing further live roots. Otherwise a small budget
+            # could repeatedly remove roots while their orphan payloads wait.
+            orphan_debt = bool(root_debris_sizes) or any(
+                canonical_segments.get((path.parent.name, path.stem)) != path
+                or segment_references.get((path.parent.name, path.stem), 0) == 0
+                for path in segment_sizes
+            ) or any(
+                canonical_chunks.get(path.stem) != path
+                or references.get(path.stem, 0) == 0
+                for path in chunk_sizes
+            )
             projected_references = references.copy()
             projected_segment_references = segment_references.copy()
             selected: list[_CapacityEntry] = []
             selected_paths: set[Path] = set()
 
             def select(entry: _CapacityEntry) -> None:
-                nonlocal projected_bytes
+                nonlocal projected_bytes, work_pending
                 if entry.path in selected_paths or (
                     entry.valid and entry.key in protected
                 ):
                     return
+                if policy.maintenance_max_deletions > 0 and orphan_debt:
+                    work_pending = True
+                    return
+                if (policy.maintenance_max_deletions > 0
+                        and len(selected) >= policy.maintenance_max_deletions):
+                    work_pending = True
+                    return
                 selected.append(entry)
                 selected_paths.add(entry.path)
                 projected_bytes -= entry.manifest_bytes
@@ -2571,6 +2611,8 @@ def select(entry: _CapacityEntry) -> None:
             affected_root_directories: set[Path] = set()
             root_debris_deleted = 0
             for path, size in root_debris_sizes.items():
+                if not admit_deletion():
+                    continue
                 try:
                     path.unlink()
                 except FileNotFoundError:
@@ -2582,6 +2624,8 @@ def select(entry: _CapacityEntry) -> None:
                     affected_root_directories.add(path.parent)
                     root_debris_deleted += size
             for entry in selected:
+                if not admit_deletion():
+                    continue
                 try:
                     entry.path.unlink()
                 except FileNotFoundError:
@@ -2627,6 +2671,8 @@ def select(entry: _CapacityEntry) -> None:
                     and remaining_segment_references.get(reference, 0) > 0
                 ):
                     continue
+                if not admit_deletion():
+                    continue
                 try:
                     path.unlink()
                 except OSError:
@@ -2654,6 +2700,8 @@ def select(entry: _CapacityEntry) -> None:
                     and remaining_references.get(path.stem, 0) > 0
                 ):
                     continue
+                if not admit_deletion():
+                    continue
                 try:
                     path.unlink()
                 except OSError:
@@ -2681,6 +2729,8 @@ def select(entry: _CapacityEntry) -> None:
             return MaintenanceReport(
                 bytes_before=bytes_before,
                 bytes_after=bytes_after,
+                deletion_attempts=deletion_attempts,
+                work_pending=work_pending,
                 bytes_reclaimed=bytes_before - bytes_after,
                 manifests_evicted=exact_removed,
                 chunks_deleted=chunks_deleted,
@@ -2694,6 +2744,9 @@ def select(entry: _CapacityEntry) -> None:
                 orphan_segments_deleted=orphan_segments_deleted,
             )
         finally:
+            self._maintenance_not_before = (
+                time.monotonic() + policy.maintenance_interval_ms / 1000.0
+            )
             guard.__exit__(None, None, None)
 
     def _manifest_path(self, identity: CacheIdentity, context_digest: str) -> Path:
diff --git a/sparkcache/persistent_context_cache/test_maintenance_budget.py b/sparkcache/persistent_context_cache/test_maintenance_budget.py
new file mode 100644
index 0000000..d613d2b
--- /dev/null
+++ b/sparkcache/persistent_context_cache/test_maintenance_budget.py
@@ -0,0 +1,175 @@
+"""Deletion pacing preserves cache references and publication barriers."""
+
+from unittest import mock
+
+import pytest
+
+from sparkcache.persistent_context_cache import cache_manifest as manifest
+from sparkcache.persistent_context_cache.cache_manifest import (
+    CapacityPolicy,
+    EntryKey,
+    ManifestStore,
+)
+from sparkcache.persistent_context_cache.test_cache_manifest import (
+    _identity,
+    _variant_chunk,
+)
+
+
+def populate(root, count=3):
+    store = ManifestStore(root)
+    identity = _identity()
+    digests = [f"{index + 1:064x}" for index in range(count)]
+    for digest in digests:
+        store.commit(
+            identity=identity,
+            context_digest=digest,
+            chunks=[_variant_chunk(digest.encode())],
+        )
+    return store, identity, digests
+
+
+def test_budget_drains_orphans_before_evicting_more_roots(tmp_path):
+    store, _, _ = populate(tmp_path)
+    policy = CapacityPolicy(
+        max_bytes=1, low_watermark_bytes=1, maintenance_max_deletions=1
+    )
+    first = store.maintain(policy)
+    assert first.manifests_evicted == first.deletion_attempts == 1
+    assert first.work_pending
+    second = store.maintain(policy)
+    assert second.manifests_evicted == 0
+    assert second.chunks_deleted == second.deletion_attempts == 1
+    for _ in range(8):
+        result = store.maintain(policy)
+        assert result.deletion_attempts <= 1
+        if result.capacity_satisfied and not result.work_pending:
+            break
+    else:
+        pytest.fail("bounded passes did not converge")
+    assert result.bytes_after == 0
+
+
+def test_budget_preserves_protected_shared_payload(tmp_path):
+    store, identity, digests = populate(tmp_path, 1)
+    sibling = "f" * 64
+    store.commit(
+        identity=identity,
+        context_digest=sibling,
+        chunks=[_variant_chunk(digests[0].encode())],
+    )
+    policy = CapacityPolicy(
+        max_bytes=1, low_watermark_bytes=1, maintenance_max_deletions=1
+    )
+    report = store.maintain(
+        policy, protected_entries=[EntryKey(identity.storage_key, sibling)]
+    )
+    assert report.manifests_evicted == 1
+    assert report.chunks_deleted == 0
+    assert not report.capacity_satisfied
+    lookup = store.lookup(identity, sibling)
+    assert lookup.is_hit
+    assert store.restore(lookup) == (_variant_chunk(digests[0].encode()),)
+    assert report.bytes_after == sum(
+        manifest._allocated_bytes(p.stat())
+        for folder in ("manifests", "chunks")
+        for p in (tmp_path / folder).rglob("*")
+        if p.is_file()
+    )
+
+
+def test_budget_counts_failed_unlink_attempts(tmp_path):
+    store, _, _ = populate(tmp_path)
+    policy = CapacityPolicy(
+        max_bytes=1, low_watermark_bytes=1, maintenance_max_deletions=1
+    )
+    with mock.patch("pathlib.Path.unlink", side_effect=OSError("busy")) as unlink:
+        report = store.maintain(policy)
+    assert unlink.call_count == report.deletion_attempts == 1
+    assert report.manifests_evicted == report.chunks_deleted == 0
+    assert report.bytes_before == report.bytes_after
+    assert report.work_pending
+
+
+def test_cooldown_skips_inventory_and_does_not_extend_itself(tmp_path):
+    store, _, _ = populate(tmp_path)
+    policy = CapacityPolicy(
+        max_bytes=1,
+        low_watermark_bytes=1,
+        maintenance_max_deletions=1,
+        maintenance_interval_ms=1000,
+    )
+    with mock.patch.object(manifest.time, "monotonic", return_value=10.0):
+        store.maintain(policy)
+    with (
+        mock.patch.object(manifest.time, "monotonic", return_value=10.9),
+        mock.patch.object(store, "_capacity_entry", side_effect=AssertionError("scan")),
+    ):
+        report = store.maintain(policy)
+    assert report.skipped_cooldown and not report.skipped_busy
+    assert not report.capacity_satisfied
+    assert report.deletion_attempts == 0
+    with mock.patch.object(manifest.time, "monotonic", return_value=11.0):
+        resumed = store.maintain(policy)
+    assert not resumed.skipped_cooldown
+    assert resumed.chunks_deleted == 1
+
+
+def test_root_barrier_failure_preserves_payload_and_starts_cooldown(tmp_path):
+    store, _, _ = populate(tmp_path, 1)
+    policy = CapacityPolicy(
+        max_bytes=1,
+        low_watermark_bytes=1,
+        maintenance_max_deletions=2,
+        maintenance_interval_ms=1000,
+    )
+    with (
+        mock.patch.object(manifest.time, "monotonic", return_value=10.0),
+        mock.patch.object(manifest, "_fsync_directory", side_effect=OSError("barrier")),
+    ):
+        with pytest.raises(OSError, match="barrier"):
+            store.maintain(policy)
+    assert len(list((tmp_path / "chunks").iterdir())) == 1
+    with mock.patch.object(manifest.time, "monotonic", return_value=10.1):
+        assert store.maintain(policy).skipped_cooldown
+
+
+@pytest.mark.parametrize(
+    "field", ["maintenance_max_deletions", "maintenance_interval_ms"]
+)
+@pytest.mark.parametrize("value", [-1, True, 1.5, "1"])
+def test_maintenance_controls_reject_invalid_values(field, value):
+    with pytest.raises(ValueError, match=field):
+        CapacityPolicy(**{field: value})
+
+
+def test_budgeted_alias_cleanup_keeps_protected_descriptor_chain(tmp_path):
+    from sparkcache.persistent_context_cache.test_prefix_aliases import _publish_source
+    from sparkcache.spark_context_cache_codec import context_prefix_digest
+
+    store, identity, tokens, _, chunks = _publish_source(
+        tmp_path, chunk_count=20, prefix_tokens=(1024, 4096)
+    )
+    digest = context_prefix_digest(tokens, identity.storage_key, token_count=4096)
+    protected = [EntryKey(identity.storage_key, digest, "prefix_alias")]
+    policy = CapacityPolicy(
+        max_bytes=1, low_watermark_bytes=1, maintenance_max_deletions=2
+    )
+    for _ in range(20):
+        report = store.maintain(policy, protected_entries=protected)
+        assert report.deletion_attempts <= 2
+        lookup = store.lookup(identity, digest, storage_mode="per_token_rows")
+        assert lookup.is_hit, lookup.reason
+        assert store.restore(lookup) == chunks[:16]
+        if report.deletion_attempts == 0:
+            break
+    else:
+        pytest.fail("cleanup did not settle around the protected alias")
+    for _ in range(30):
+        report = store.maintain(policy)
+        assert report.deletion_attempts <= 2
+        if report.capacity_satisfied and not report.work_pending:
+            break
+    else:
+        pytest.fail("descriptor and payload cleanup did not converge")
+    assert report.bytes_after == 0
diff --git a/sparkcache/spark_context_cache_config.py b/sparkcache/spark_context_cache_config.py
index 2d57eb4..e18889a 100644
--- a/sparkcache/spark_context_cache_config.py
+++ b/sparkcache/spark_context_cache_config.py
@@ -634,6 +634,16 @@ def parse_connector_config(
             max_bytes=max_bytes,
             low_watermark_bytes=low_watermark_bytes,
             ttl_seconds=ttl_seconds,
+            maintenance_max_deletions=_nonnegative_config_int(
+                extra("spark_cache_maintenance_max_deletions", os.environ.get(
+                    "SPARK_CONTEXT_CACHE_MAINTENANCE_MAX_DELETIONS", "0")),
+                "spark_cache_maintenance_max_deletions",
+            ),
+            maintenance_interval_ms=_nonnegative_config_int(
+                extra("spark_cache_maintenance_interval_ms", os.environ.get(
+                    "SPARK_CONTEXT_CACHE_MAINTENANCE_INTERVAL_MS", "0")),
+                "spark_cache_maintenance_interval_ms",
+            ),
         )
     except ValueError as error:
         raise RuntimeError(f"spark-context-cache: {error}") from error
diff --git a/sparkcache/spark_context_cache_connector.py b/sparkcache/spark_context_cache_connector.py
index b390718..07ac47a 100644
--- a/sparkcache/spark_context_cache_connector.py
+++ b/sparkcache/spark_context_cache_connector.py
@@ -616,6 +616,10 @@ def reduce(self) -> dict[str, int | float]:
                 ),
             }
             reduced.update({key: value for key, value in alerts.items() if value})
+            for name in ("deletion_attempts", "budget_exhausted", "skipped_cooldown"):
+                reduced[f"sparkcache_maintenance_{name}"] = sum(
+                    int(status.get(f"maintenance_{name}", 0)) for status in capacity
+                )
         async_capture = [
             report.get("async_capture")
             for report in reports
@@ -794,6 +798,21 @@ class SparkCachePromMetrics(KVConnectorPromMetrics):
             "Physical ranks reporting an active capacity scan or reconciliation at their last worker reports.",
             1.0,
         ),
+        "sparkcache_maintenance_deletion_attempts": (
+            "vllm:sparkcache_maintenance_deletion_attempts",
+            "Reported cumulative unlink attempts summed across physical ranks; resets with workers.",
+            1.0,
+        ),
+        "sparkcache_maintenance_budget_exhausted": (
+            "vllm:sparkcache_maintenance_budget_exhausted",
+            "Reported cumulative passes deferring deletion work across physical ranks.",
+            1.0,
+        ),
+        "sparkcache_maintenance_skipped_cooldown": (
+            "vllm:sparkcache_maintenance_skipped_cooldown",
+            "Reported cumulative cooldown skips across physical ranks.",
+            1.0,
+        ),
         "sparkcache_capture_delayed_requests": (
             "vllm:sparkcache_capture_delayed_requests",
             "Maximum delayed SparkCache capture requests on any physical rank.",
@@ -3076,7 +3095,7 @@ def register_kv_caches(self, kv_caches: dict[str, torch.Tensor]) -> None:
         ):
             report = self._maintain_capacity(force=True)
             self._ensure_capacity_thread()
-            if (report is not None and report.skipped_busy) or not bool(
+            if (report is not None and (report.skipped_busy or report.skipped_cooldown)) or not bool(
                 self._capacity_status["capacity_satisfied"]
             ):
                 self._capacity_wakeup.set()
@@ -3352,6 +3371,13 @@ def _perform_capacity_maintenance_locked(
                 "spark-context-cache: capacity maintenance failed: %s", error
             )
             return None
+        if report.skipped_cooldown:
+            self.counters["capacity_skipped_cooldown"] = (
+                self.counters.get("capacity_skipped_cooldown", 0) + 1
+            )
+            if wake_worker_on_unsatisfied:
+                self._capacity_wakeup.set()
+            return report
         if report.skipped_busy:
             self.counters["capacity_skipped_busy"] += 1
             self._capacity_status.update(
@@ -3369,6 +3395,12 @@ def _perform_capacity_maintenance_locked(
             bytes_exact=True,
             capacity_satisfied=report.capacity_satisfied,
         )
+        self.counters["capacity_deletion_attempts"] = (
+            self.counters.get("capacity_deletion_attempts", 0) + report.deletion_attempts
+        )
+        self.counters["capacity_budget_exhausted"] = (
+            self.counters.get("capacity_budget_exhausted", 0) + int(report.work_pending)
+        )
         self.counters["capacity_runs"] += 1
         self.counters["capacity_manifests_evicted"] += report.manifests_evicted
         self.counters["capacity_chunks_deleted"] += report.chunks_deleted
@@ -3426,12 +3458,13 @@ def _perform_capacity_maintenance_locked(
         # alias. The targeted checks above retain the offer when either root
         # remains; this complete pass also catches entries removed as debris.
         self._reconcile_held_capacity()
-        if not report.capacity_satisfied and wake_worker_on_unsatisfied:
+        if (not report.capacity_satisfied or report.work_pending) and wake_worker_on_unsatisfied:
             self._capacity_wakeup.set()
         if force or report.bytes_reclaimed:
             logger.info(
                 "spark-context-cache: capacity bytes=%d max=%d reclaimed=%d"
-                " manifests=%d chunks=%d orphans=%d satisfied=%s",
+                " manifests=%d chunks=%d orphans=%d satisfied=%s"
+                " deletion_attempts=%d work_pending=%s",
                 report.bytes_after,
                 policy.max_bytes,
                 report.bytes_reclaimed,
@@ -3439,6 +3472,8 @@ def _perform_capacity_maintenance_locked(
                 report.chunks_deleted,
                 report.orphan_chunks_deleted,
                 report.capacity_satisfied,
+                report.deletion_attempts,
+                report.work_pending,
             )
         return report
 
@@ -3463,7 +3498,7 @@ def _post_commit_was_evicted_locked(
             force=force_maintenance,
             wake_worker_on_unsatisfied=True,
         )
-        if report is not None and not report.skipped_busy:
+        if report is not None and not (report.skipped_busy or report.skipped_cooldown):
             exact_evicted = (
                 EntryKey(
                     identity.storage_key,
@@ -3691,7 +3726,7 @@ def _finalize_streaming_capacity_commits(
                 report = self._maintain_capacity_locked(force=True)
                 if (
                     report is None
-                    or report.skipped_busy
+                    or (report.skipped_busy or report.skipped_cooldown)
                     or not report.capacity_satisfied
                 ):
                     return False
@@ -3816,6 +3851,7 @@ def _capacity_worker_main(self) -> None:
             retry_unsatisfied = bool(
                 report is None
                 or not report.capacity_satisfied
+                or report.work_pending
                 or not bool(self._capacity_status["capacity_satisfied"])
             )
             if retry_unsatisfied:
@@ -6510,6 +6546,9 @@ def get_kv_connector_stats(self):
                     "streaming_capacity_shutdown_dropped"
                 ],
                 maintenance_retries=self.counters["capacity_retries"],
+                maintenance_deletion_attempts=self.counters.get("capacity_deletion_attempts", 0),
+                maintenance_budget_exhausted=self.counters.get("capacity_budget_exhausted", 0),
+                maintenance_skipped_cooldown=self.counters.get("capacity_skipped_cooldown", 0),
             )
             report["capacity"] = capacity
         runtime = self._async_page_capture_runtime
diff --git a/sparkcache/test_maintenance_pacing.py b/sparkcache/test_maintenance_pacing.py
new file mode 100644
index 0000000..a261f33
--- /dev/null
+++ b/sparkcache/test_maintenance_pacing.py
@@ -0,0 +1,61 @@
+"""Forced maintenance respects cooldown without repeating survivor scans."""
+
+from unittest import mock
+
+from sparkcache import test_spark_context_cache_connector as fixtures
+from sparkcache.persistent_context_cache.cache_manifest import (
+    CapacityPolicy,
+    MaintenanceReport,
+)
+
+
+def test_forced_post_commit_cooldown_does_not_reconcile_inventory(tmp_path):
+    connector = fixtures.AsyncRestoreTests()._cohort_connector(tmp_path)
+    connector._capacity_policy = CapacityPolicy(
+        max_bytes=10, low_watermark_bytes=8, maintenance_interval_ms=1000
+    )
+    connector._capacity_estimated_bytes = 11
+    connector._capacity_status["capacity_satisfied"] = False
+    try:
+        with (
+            mock.patch.object(
+                connector._store,
+                "maintain",
+                return_value=MaintenanceReport(
+                    capacity_satisfied=False, skipped_cooldown=True
+                ),
+            ),
+            mock.patch.object(connector, "_reconcile_held_capacity") as reconcile,
+        ):
+            report = connector._maintain_capacity(force=True)
+        assert report.skipped_cooldown
+        assert not connector._capacity_status["capacity_satisfied"]
+        assert connector._capacity_estimated_bytes == 11
+        assert connector.counters["capacity_skipped_cooldown"] == 1
+        reconcile.assert_not_called()
+    finally:
+        connector.shutdown()
+
+
+def test_pacing_totals_aggregate_physical_ranks():
+    from sparkcache.spark_context_cache_connector import SparkCacheStats
+
+    stats = SparkCacheStats(
+        data={
+            "reports": [
+                {
+                    "rank": rank,
+                    "capacity": {
+                        "maintenance_deletion_attempts": 128,
+                        "maintenance_budget_exhausted": rank,
+                        "maintenance_skipped_cooldown": 2,
+                    },
+                }
+                for rank in range(4)
+            ]
+        }
+    )
+    reduced = stats.reduce()
+    assert reduced["sparkcache_maintenance_deletion_attempts"] == 512
+    assert reduced["sparkcache_maintenance_budget_exhausted"] == 6
+    assert reduced["sparkcache_maintenance_skipped_cooldown"] == 8
diff --git a/sparkcache/test_spark_context_cache_config.py b/sparkcache/test_spark_context_cache_config.py
index 1e3ce8f..86baf6f 100644
--- a/sparkcache/test_spark_context_cache_config.py
+++ b/sparkcache/test_spark_context_cache_config.py
@@ -297,6 +297,8 @@ def test_capacity_policy_from_env(self) -> None:
                 "SPARK_CONTEXT_CACHE_MAX_BYTES": "1000000",
                 "SPARK_CONTEXT_CACHE_LOW_WATERMARK_BYTES": "900000",
                 "SPARK_CONTEXT_CACHE_TTL_SECONDS": "3600",
+                "SPARK_CONTEXT_CACHE_MAINTENANCE_MAX_DELETIONS": "128",
+                "SPARK_CONTEXT_CACHE_MAINTENANCE_INTERVAL_MS": "1000",
                 "SPARK_CONTEXT_CACHE_MODEL_PROFILE": "glm52-nvfp4",
                 "SPARK_CONTEXT_CACHE_TARGET_CHECKPOINT_SHA256": "1" * 64,
                 "SPARK_CONTEXT_CACHE_DRAFT_CHECKPOINT_SHA256": "2" * 64,
@@ -307,6 +309,8 @@ def test_capacity_policy_from_env(self) -> None:
         self.assertEqual(config.capacity_policy.max_bytes, 1000000)
         self.assertEqual(config.capacity_policy.low_watermark_bytes, 900000)
         self.assertEqual(config.capacity_policy.ttl_seconds, 3600)
+        self.assertEqual(config.capacity_policy.maintenance_max_deletions, 128)
+        self.assertEqual(config.capacity_policy.maintenance_interval_ms, 1000)
 
     def test_missing_model_profile_is_rejected_explicitly(self) -> None:
         vllm, _ = _make_vllm_config()

From 78ca0bdc5d63629ef6ee6239546921898956fa2e Mon Sep 17 00:00:00 2001
From: FujitsuPolycom <87842395+FujitsuPolycom@users.noreply.github.com>
Date: Sun, 6 Sep 2026 17:49:58 -0500
Subject: [PATCH 3/7] Retain low-watermark reclamation targets across paced
 passes

Keep a process-local pressure target until bounded cleanup reaches the low watermark. Continue background retries and non-forced maintenance below the high watermark while work remains, including after streaming commit finalization. Disabled deletion budgets preserve existing behavior and persisted cache identities are unchanged.

Validation: 1251 SparkCache and deployment tests passed with 9 skips. Budget sizes 1, 2, and 4 converge to the same reclaimed bytes as unrestricted cleanup at 99/80 percent watermarks.
---
 deploy/deepseek_v4/tp4_profile.json           |  2 +-
 deploy/glm52_35bpw/profile.json               |  2 +-
 sparkcache/README.md                          |  4 +++
 .../cache_manifest.py                         | 19 ++++++++--
 .../test_maintenance_budget.py                | 35 +++++++++++++++++++
 sparkcache/spark_context_cache_connector.py   |  8 +++--
 sparkcache/test_maintenance_pacing.py         | 20 +++++++++++
 7 files changed, 84 insertions(+), 6 deletions(-)

diff --git a/deploy/deepseek_v4/tp4_profile.json b/deploy/deepseek_v4/tp4_profile.json
index 7e25529..82c6110 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": "6ecc20725a23c4d260cf227b1d2805fe925d3def471caa0ad7b33267f704da76"
+    "source_sha256": "3aab8ebd902e0bff0752b83af775d5c74db34c20bc5352bec451a5b6e7fc652d"
   },
   "model": {
     "repository": "deepseek-ai/DeepSeek-V4-Flash-0731",
diff --git a/deploy/glm52_35bpw/profile.json b/deploy/glm52_35bpw/profile.json
index d71763a..98c0dec 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": "6ecc20725a23c4d260cf227b1d2805fe925d3def471caa0ad7b33267f704da76"
+    "source_sha256": "3aab8ebd902e0bff0752b83af775d5c74db34c20bc5352bec451a5b6e7fc652d"
   },
   "model": {
     "repository": "brandonmusic/GLM-5.2-EXL3-TR3v4-3.5bpw-MTP78",
diff --git a/sparkcache/README.md b/sparkcache/README.md
index 69fb344..b39c470 100644
--- a/sparkcache/README.md
+++ b/sparkcache/README.md
@@ -336,6 +336,10 @@ A pass can stop above the low watermark
 or the capacity maximum; optional store admission remains blocked while
 capacity is unsatisfied.
 
+A process-local pressure target retains the low watermark across passes,
+even after usage falls below the high watermark. Restarting the process
+loses this target; a later high-watermark crossing establishes it again.
+
 Background retries complete deferred cleanup without
 making serving wait. Root-directory durability barriers still precede object
 removal, and protected publication roots retain their complete object graphs.
diff --git a/sparkcache/persistent_context_cache/cache_manifest.py b/sparkcache/persistent_context_cache/cache_manifest.py
index 45a87ca..d57c3b3 100644
--- a/sparkcache/persistent_context_cache/cache_manifest.py
+++ b/sparkcache/persistent_context_cache/cache_manifest.py
@@ -2528,8 +2528,18 @@ def root_files(root: Path) -> tuple[Path, ...]:
                 + sum(segment_sizes.values())
                 + sum(chunk_sizes.values())
             )
-            pressure_triggered = (
-                policy.max_bytes > 0 and bytes_before > policy.max_bytes
+            pressure_target = (policy.max_bytes, policy.low_watermark_bytes)
+            retained_target = getattr(self, "_maintenance_pressure_target", None)
+            pressure_triggered = policy.max_bytes > 0 and (
+                bytes_before > policy.max_bytes
+                or (policy.maintenance_max_deletions > 0
+                    and retained_target == pressure_target
+                    and bytes_before > policy.low_watermark_bytes)
+            )
+            # Preserve high-to-low hysteresis across bounded passes. Falling
+            # below the high watermark alone does not complete reclamation.
+            self._maintenance_pressure_target = (
+                pressure_target if pressure_triggered and policy.maintenance_max_deletions > 0 else None
             )
             projected_bytes = (
                 bytes_before
@@ -2722,6 +2732,11 @@ def select(entry: _CapacityEntry) -> None:
                 - segment_bytes_deleted
                 - chunk_bytes_deleted,
             )
+            if (pressure_triggered and policy.maintenance_max_deletions > 0
+                    and bytes_after > policy.low_watermark_bytes):
+                work_pending = True
+            else:
+                self._maintenance_pressure_target = None
             exact_removed = sum(entry.key.root_kind == "manifest" for entry in removed)
             aliases_removed = sum(
                 entry.key.root_kind == "prefix_alias" for entry in removed
diff --git a/sparkcache/persistent_context_cache/test_maintenance_budget.py b/sparkcache/persistent_context_cache/test_maintenance_budget.py
index d613d2b..53c8ff9 100644
--- a/sparkcache/persistent_context_cache/test_maintenance_budget.py
+++ b/sparkcache/persistent_context_cache/test_maintenance_budget.py
@@ -173,3 +173,38 @@ def test_budgeted_alias_cleanup_keeps_protected_descriptor_chain(tmp_path):
     else:
         pytest.fail("descriptor and payload cleanup did not converge")
     assert report.bytes_after == 0
+
+
+@pytest.mark.parametrize("budget", [1, 2, 4])
+def test_pressure_target_survives_crossing_high_watermark(tmp_path, budget):
+    import shutil
+
+    seed = tmp_path / "seed"
+    store, _, _ = populate(seed, 20)
+    before = store.maintain(
+        CapacityPolicy(max_bytes=10**9, low_watermark_bytes=10**9)
+    ).bytes_after
+    high = before * 99 // 100
+    low = before * 80 // 100
+    baseline_path = tmp_path / "baseline"
+    shutil.copytree(seed, baseline_path)
+    baseline = ManifestStore(baseline_path).maintain(
+        CapacityPolicy(max_bytes=high, low_watermark_bytes=low)
+    )
+    policy = CapacityPolicy(
+        max_bytes=high, low_watermark_bytes=low, maintenance_max_deletions=budget
+    )
+    saw_pending_below_high = False
+    for _ in range(100):
+        report = store.maintain(policy)
+        saw_pending_below_high |= (
+            low < report.bytes_after <= high and report.work_pending
+        )
+        if report.capacity_satisfied and not report.work_pending:
+            break
+    else:
+        pytest.fail("maintenance did not reach its retained low watermark")
+    assert saw_pending_below_high
+    assert report.bytes_after == baseline.bytes_after
+    assert report.bytes_after <= low
+    assert store._maintenance_pressure_target is None
diff --git a/sparkcache/spark_context_cache_connector.py b/sparkcache/spark_context_cache_connector.py
index 07ac47a..126aaf0 100644
--- a/sparkcache/spark_context_cache_connector.py
+++ b/sparkcache/spark_context_cache_connector.py
@@ -3328,7 +3328,8 @@ def _maintain_capacity_locked(
         if not policy.enabled:
             return None
         if not force and (
-            policy.max_bytes == 0 or self._capacity_estimated_bytes <= policy.max_bytes
+            (policy.max_bytes == 0 or self._capacity_estimated_bytes <= policy.max_bytes)
+            and not self._capacity_status.get("maintenance_work_pending", False)
         ):
             return None
         self._capacity_maintenance_depth = getattr(self, "_capacity_maintenance_depth", 0) + 1
@@ -3394,6 +3395,7 @@ def _perform_capacity_maintenance_locked(
             bytes=report.bytes_after,
             bytes_exact=True,
             capacity_satisfied=report.capacity_satisfied,
+            maintenance_work_pending=report.work_pending,
         )
         self.counters["capacity_deletion_attempts"] = (
             self.counters.get("capacity_deletion_attempts", 0) + report.deletion_attempts
@@ -3838,7 +3840,9 @@ def _capacity_worker_main(self) -> None:
                     with self._capacity_handoff_cv:
                         self._streaming_capacity_pending.difference_update(resolved)
                         self._capacity_handoff_cv.notify_all()
-                    retry_unsatisfied = False
+                    retry_unsatisfied = bool(
+                        self._capacity_status.get("maintenance_work_pending", False)
+                    )
                 else:
                     self.counters["streaming_capacity_retries"] += 1
                     retry_unsatisfied = True
diff --git a/sparkcache/test_maintenance_pacing.py b/sparkcache/test_maintenance_pacing.py
index a261f33..3cca4b4 100644
--- a/sparkcache/test_maintenance_pacing.py
+++ b/sparkcache/test_maintenance_pacing.py
@@ -59,3 +59,23 @@ def test_pacing_totals_aggregate_physical_ranks():
     assert reduced["sparkcache_maintenance_deletion_attempts"] == 512
     assert reduced["sparkcache_maintenance_budget_exhausted"] == 6
     assert reduced["sparkcache_maintenance_skipped_cooldown"] == 8
+
+
+def test_latched_pressure_runs_maintenance_below_high_watermark(tmp_path):
+    connector = fixtures.AsyncRestoreTests()._cohort_connector(tmp_path)
+    connector._capacity_policy = CapacityPolicy(
+        max_bytes=10, low_watermark_bytes=8, maintenance_max_deletions=1
+    )
+    connector._capacity_estimated_bytes = 9
+    connector._capacity_status["maintenance_work_pending"] = True
+    try:
+        with mock.patch.object(
+            connector,
+            "_perform_capacity_maintenance_locked",
+            return_value=MaintenanceReport(work_pending=True),
+        ) as run:
+            report = connector._maintain_capacity(force=False)
+        run.assert_called_once()
+        assert report.work_pending
+    finally:
+        connector.shutdown()

From 7fc9f85509624a0f9c5bc6bd7c5f70f1a43d389e Mon Sep 17 00:00:00 2001
From: FujitsuPolycom <87842395+FujitsuPolycom@users.noreply.github.com>
Date: Sun, 6 Sep 2026 18:00:16 -0500
Subject: [PATCH 4/7] Compare canonical path strings during maintenance
 ordering

Compute inventory ordering keys once instead of repeatedly normalizing Path components during comparisons. Canonical namespace and digest ordering remains deterministic; stored identities and namespaces do not change. Validation: 1,156 CPU tests passed with eight skips, Ruff passed, and Linux profiling removed Path comparison calls without establishing a single-pass latency improvement.
---
 deploy/deepseek_v4/tp4_profile.json           |  2 +-
 deploy/glm52_35bpw/profile.json               |  2 +-
 .../cache_manifest.py                         | 21 ++++++++++++-------
 3 files changed, 15 insertions(+), 10 deletions(-)

diff --git a/deploy/deepseek_v4/tp4_profile.json b/deploy/deepseek_v4/tp4_profile.json
index 82c6110..4e837b5 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": "3aab8ebd902e0bff0752b83af775d5c74db34c20bc5352bec451a5b6e7fc652d"
+    "source_sha256": "e4a3451b2849cff6e7b8dea32a741ce73698c4585c530db22457088d99496372"
   },
   "model": {
     "repository": "deepseek-ai/DeepSeek-V4-Flash-0731",
diff --git a/deploy/glm52_35bpw/profile.json b/deploy/glm52_35bpw/profile.json
index 98c0dec..bf797f7 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": "3aab8ebd902e0bff0752b83af775d5c74db34c20bc5352bec451a5b6e7fc652d"
+    "source_sha256": "e4a3451b2849cff6e7b8dea32a741ce73698c4585c530db22457088d99496372"
   },
   "model": {
     "repository": "brandonmusic/GLM-5.2-EXL3-TR3v4-3.5bpw-MTP78",
diff --git a/sparkcache/persistent_context_cache/cache_manifest.py b/sparkcache/persistent_context_cache/cache_manifest.py
index d57c3b3..ba00050 100644
--- a/sparkcache/persistent_context_cache/cache_manifest.py
+++ b/sparkcache/persistent_context_cache/cache_manifest.py
@@ -2435,12 +2435,12 @@ def admit_deletion() -> bool:
             manifests_root = self.root / "manifests"
             aliases_root = self.root / "prefix-aliases"
             manifest_paths = (
-                tuple(sorted(manifests_root.glob("*/*.json")))
+                tuple(sorted(manifests_root.glob("*/*.json"), key=os.fspath))
                 if manifests_root.is_dir()
                 else ()
             )
             alias_paths = (
-                tuple(sorted(aliases_root.glob("*/*.json")))
+                tuple(sorted(aliases_root.glob("*/*.json"), key=os.fspath))
                 if aliases_root.is_dir()
                 else ()
             )
@@ -2461,11 +2461,12 @@ def root_files(root: Path) -> tuple[Path, ...]:
                     return ()
                 return tuple(
                     sorted(
-                        path
+                        (path
                         for directory in root.iterdir()
                         if directory.is_dir()
                         for path in directory.iterdir()
-                        if path.is_file()
+                        if path.is_file()),
+                        key=os.fspath,
                     )
                 )
 
@@ -2500,7 +2501,8 @@ def root_files(root: Path) -> tuple[Path, ...]:
             chunk_directory = self.root / "chunks"
             chunk_paths = (
                 tuple(
-                    sorted(path for path in chunk_directory.iterdir() if path.is_file())
+                    sorted((path for path in chunk_directory.iterdir() if path.is_file()),
+                           key=os.fspath)
                 )
                 if chunk_directory.is_dir()
                 else ()
@@ -2604,7 +2606,10 @@ def select(entry: _CapacityEntry) -> None:
                         if path is not None:
                             projected_bytes -= segment_sizes.get(path, 0)
 
-            ordered = sorted(entries, key=lambda entry: (entry.mtime_ns, entry.path))
+            # Cache namespaces and digests have fixed-width canonical names.
+            # Compare each path's string once instead of normalizing Path
+            # components throughout every comparison in the inventory sort.
+            ordered = sorted(entries, key=lambda entry: (entry.mtime_ns, os.fspath(entry.path)))
             for entry in ordered:
                 expired = policy.ttl_seconds > 0 and (
                     current_ns - entry.mtime_ns >= policy.ttl_seconds * 10**9
@@ -2649,7 +2654,7 @@ def select(entry: _CapacityEntry) -> None:
             # Root removals are durable before any object they authorized can
             # be collected. A failed root-directory barrier stops maintenance
             # with every shared segment and chunk still present.
-            for directory in sorted(affected_root_directories):
+            for directory in sorted(affected_root_directories, key=os.fspath):
                 _fsync_directory(directory)
 
             remaining_references: Counter[str] = Counter()
@@ -2692,7 +2697,7 @@ def select(entry: _CapacityEntry) -> None:
                 affected_segment_directories.add(path.parent)
                 if path in initial_orphan_segments:
                     orphan_segments_deleted += 1
-            for directory in sorted(affected_segment_directories):
+            for directory in sorted(affected_segment_directories, key=os.fspath):
                 _fsync_directory(directory)
 
             chunks_deleted = 0

From 11658f0a3b9f155af1d95ff78a807c9b374f6093 Mon Sep 17 00:00:00 2001
From: FujitsuPolycom <87842395+FujitsuPolycom@users.noreply.github.com>
Date: Sun, 6 Sep 2026 18:02:35 -0500
Subject: [PATCH 5/7] Reuse the guarded maintenance inventory to reconcile
 cache offers

Qualify surviving roots from validated descriptors and unique-file sizes during the exclusive pass, avoiding repeated root reads and shared-chunk probes. Preserve exact-root precedence and defer stale-inventory withdrawals when offers change, including same-digest republication. Restore integrity checks, cache namespaces, and persisted formats are unchanged.

CPU validation: 1279 tests passed with 9 skips; Ruff clean. A Windows NTFS fixture with 257 flat-page roots reduced complete maintenance/reconciliation median latency from 175.23 ms to 72.22 ms and chunk metadata probes from 4995 to 514. GPU, serving, and NVMe qualification remains pending.
---
 deploy/deepseek_v4/tp4_profile.json           |   2 +-
 deploy/glm52_35bpw/profile.json               |   2 +-
 docs/maintenance-inventory-validation.json    |  68 +++++++
 docs/maintenance-inventory-validation.md      |  49 +++++
 sparkcache/README.md                          |  29 +++
 .../cache_manifest.py                         |  24 ++-
 sparkcache/spark_context_cache_connector.py   |  86 ++++----
 sparkcache/test_maintenance_survivors.py      | 186 ++++++++++++++++++
 .../test_spark_context_cache_connector.py     |   1 +
 tools/benchmark_maintenance_inventory.py      | 108 ++++++++++
 10 files changed, 500 insertions(+), 55 deletions(-)
 create mode 100644 docs/maintenance-inventory-validation.json
 create mode 100644 docs/maintenance-inventory-validation.md
 create mode 100644 sparkcache/test_maintenance_survivors.py
 create mode 100644 tools/benchmark_maintenance_inventory.py

diff --git a/deploy/deepseek_v4/tp4_profile.json b/deploy/deepseek_v4/tp4_profile.json
index 4e837b5..a0cb9d7 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": "e4a3451b2849cff6e7b8dea32a741ce73698c4585c530db22457088d99496372"
+    "source_sha256": "7efffbec85cce754b7aad057cead39f4b1e1cb7f35833e8df8e487d3d4a46867"
   },
   "model": {
     "repository": "deepseek-ai/DeepSeek-V4-Flash-0731",
diff --git a/deploy/glm52_35bpw/profile.json b/deploy/glm52_35bpw/profile.json
index bf797f7..08a6250 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": "e4a3451b2849cff6e7b8dea32a741ce73698c4585c530db22457088d99496372"
+    "source_sha256": "7efffbec85cce754b7aad057cead39f4b1e1cb7f35833e8df8e487d3d4a46867"
   },
   "model": {
     "repository": "brandonmusic/GLM-5.2-EXL3-TR3v4-3.5bpw-MTP78",
diff --git a/docs/maintenance-inventory-validation.json b/docs/maintenance-inventory-validation.json
new file mode 100644
index 0000000..8266459
--- /dev/null
+++ b/docs/maintenance-inventory-validation.json
@@ -0,0 +1,68 @@
+{
+  "schema": "sparkcache-maintenance-inventory-benchmark/v1",
+  "status": "research-only",
+  "conditions": {
+    "platform": "Windows-11-10.0.26200-SP0",
+    "filesystem": "NTFS",
+    "filesystem_verification": "Get-Volume -DriveLetter C",
+    "python_execution": "native Windows",
+    "paired_method": "Separate sequential processes constructed equivalent temporary caches and invoked SparkContextCacheConnector._maintain_capacity(force=True).",
+    "timing": "Median of seven complete maintenance/reconciliation passes after an untimed instrumented pass; garbage collection precedes each timer.",
+    "excluded": ["fixture construction", "publication", "Python imports", "I/O counting instrumentation"],
+    "included": ["exclusive cache lock", "complete reference inventory", "offer reconciliation"],
+    "eviction": "disabled by a 1000000000-byte capacity ceiling",
+    "gpu_execution": false,
+    "serving_or_nvme_qualification": false,
+    "tool": "tools/benchmark_maintenance_inventory.py"
+  },
+  "sources": {
+    "baseline": {
+      "revision": "78ca0bdc5d63629ef6ee6239546921898956fa2e",
+      "cache_manifest_sha256_lf": "88caa5b104407a212b4344b82ab914d6c5c7e72d021976ba6d1c44d13f7f2f2e",
+      "connector_sha256_lf": "082f2820f45cdf951629d045b5d316461f86372fae70425be9b5903825395a97"
+    },
+    "survivor_inventory": {
+      "mechanism": "MaintenanceReport.surviving_entries supplies metadata-qualified offers from the exclusive inventory pass.",
+      "cache_manifest_sha256_lf": "ddb3e2e00808791ed585122b8ab1171ac7db391c19fc62151f0f39107dee256a",
+      "connector_sha256_lf": "f92b1fcaf3385e5d4487b246f06c1de3fb02e8f05b67c055ab760d91ccd86eb4"
+    }
+  },
+  "flat_page_history": {
+    "layout": "Opaque 64-byte attention pages grouped by 256 tokens, plus a 32-byte recurrent-state page.",
+    "branches": 8,
+    "extensions_per_branch": 32,
+    "shared_initial_tokens": 512,
+    "extension_tokens": 256,
+    "roots": 257,
+    "unique_chunks": 257,
+    "chunk_references": 4481,
+    "baseline": {
+      "root_reads": 514,
+      "chunk_stats": 4995,
+      "elapsed_ms": [185.0543999898946, 183.6131000018213, 173.09400001249742, 178.17310000828002, 174.59690000396222, 174.0127999946708, 175.22859999735374],
+      "median_ms": 175.22859999735374
+    },
+    "survivor_inventory": {
+      "root_reads": 257,
+      "chunk_stats": 514,
+      "elapsed_ms": [72.49789999332279, 70.9155000076862, 72.2200000018347, 70.33850000880193, 76.28729999123607, 89.75840000493918, 72.21110000682529],
+      "median_ms": 72.2200000018347
+    },
+    "conclusion": "Complete pass latency decreased 58.8 percent in this CPU fixture. Each root was read once; shared-chunk metadata probes scaled with unique files rather than root references. This does not establish a serving-throughput improvement."
+  },
+  "token_rows": {
+    "roots": 64,
+    "unique_chunks": 32,
+    "chunk_references": 2048,
+    "baseline": {
+      "root_reads": 128, "chunk_stats": 2112,
+      "elapsed_ms": [37.09349999553524, 37.26160000951495, 37.37550000369083, 37.524400002439506, 37.23980000359006, 41.24779999256134, 52.597900008549914],
+      "median_ms": 37.37550000369083
+    },
+    "survivor_inventory": {
+      "root_reads": 64, "chunk_stats": 64,
+      "elapsed_ms": [10.044100010418333, 10.23929999792017, 10.25409999419935, 10.973099997499958, 10.138300000107847, 9.93929999822285, 9.973099993658252],
+      "median_ms": 10.138300000107847
+    }
+  }
+}
diff --git a/docs/maintenance-inventory-validation.md b/docs/maintenance-inventory-validation.md
new file mode 100644
index 0000000..5191be6
--- /dev/null
+++ b/docs/maintenance-inventory-validation.md
@@ -0,0 +1,49 @@
+# Maintenance inventory and offer reconciliation
+
+Status: **implemented**. Serving-performance qualification is **research-only**.
+
+`MaintenanceReport.surviving_entries` identifies metadata-qualified roots from
+the exclusive inventory pass. Offer reconciliation reuses that result without
+rereading roots or probing shared payload files once per reference.
+
+The [CPU benchmark record](maintenance-inventory-validation.json) measures the
+complete maintenance and reconciliation call on Windows 11 with NTFS. It
+compares equivalent fixtures against source revision `78ca0bd`.
+
+| Fixture | Baseline median | Inventory median | Root reads | Chunk metadata probes |
+|---|---:|---:|---:|---:|
+| Eight branches, 32 flat page extensions each | 175.23 ms | 72.22 ms | 514 to 257 | 4,995 to 514 |
+| 64 token-row roots sharing 32 chunks | 37.38 ms | 10.14 ms | 128 to 64 | 2,112 to 64 |
+
+The page fixture includes attention and recurrent-state bytes. It exercises
+257 roots and 4,481 chunk references with tiny payloads, not a loaded model.
+The measurements do not qualify DGX serving throughput or NVMe behavior.
+
+Each normal pass still reads the complete inventory and traverses its reference
+graph. Root parsing falls from two passes to one, and chunk metadata probes
+scale with unique files instead of the total number of root references.
+
+Run the [benchmark tool](../tools/benchmark_maintenance_inventory.py) from the
+checkout being measured. Use the same tool against each source checkout.
+Install the repository's CPU test dependencies; the connector fixture stubs
+its model-runtime interfaces.
+
+```powershell
+$env:BENCH_FILESYSTEM = 'NTFS'
+$env:BENCH_KIND = 'pages'
+python tools/benchmark_maintenance_inventory.py 8 32
+$env:BENCH_KIND = 'rows'
+python tools/benchmark_maintenance_inventory.py 64 32
+```
+
+Set `BENCH_FILESYSTEM` to the independently verified filesystem. Fixture
+construction and counting instrumentation are outside the seven timed passes.
+The artifact records normalized source hashes for both measured implementations.
+
+Regression tests in `sparkcache/test_maintenance_survivors.py` cover shared
+payload metadata, corrupt exact roots shadowing aliases, storage-mode
+eligibility, concurrent publication, and repeated inventory freshness.
+
+The snapshot does not authorize cache restoration. Payload integrity remains
+mandatory at restore, and same-sized corruption becomes a verified miss.
+No cache identity, namespace, or persisted format changes.
diff --git a/sparkcache/README.md b/sparkcache/README.md
index b39c470..d06b1e6 100644
--- a/sparkcache/README.md
+++ b/sparkcache/README.md
@@ -347,6 +347,35 @@ removal, and protected publication roots retain their complete object graphs.
 This is **not a scan-size or wall-clock bound**. Each admitted pass authenticates
 the complete reference inventory and reconciles survivors.
 
+Status: **implemented**. A completed pass also returns metadata-qualified
+surviving roots while the exclusive filesystem guard is held.
+
+The connector
+uses this inventory to reconcile offers without rereading every root or
+restatting shared chunks once per reference.
+
+Qualification checks logical
+file sizes, not allocated disk space. Exact roots shadow aliases even when
+invalid; aliases are eligible only for token-row storage.
+
+Protected roots and
+their referenced objects retain the same deletion rules.
+
+The inventory is not retained between passes. Failed or busy passes use
+independent metadata probes.
+
+If the offered-digest inventory changes during a
+pass or probe, reconciliation defers withdrawals to a stable pass, preserving
+concurrent publication of the same digest.
+
+The connector counter
+`capacity_stale_inventory_snapshots` and worker capacity-report field
+`maintenance_stale_inventory_snapshots` count these deferrals.
+
+Restore still
+authenticates payload bytes; metadata qualification cannot establish their
+integrity or turn an invalid restore into a hit.
+
 One filesystem
 operation or durability barrier can take arbitrarily long. Smaller deletion
 budgets can increase total scan work.
diff --git a/sparkcache/persistent_context_cache/cache_manifest.py b/sparkcache/persistent_context_cache/cache_manifest.py
index ba00050..0c669b7 100644
--- a/sparkcache/persistent_context_cache/cache_manifest.py
+++ b/sparkcache/persistent_context_cache/cache_manifest.py
@@ -482,6 +482,9 @@ class MaintenanceReport:
     aliases_evicted: int = 0
     segments_deleted: int = 0
     orphan_segments_deleted: int = 0
+    # Metadata-qualified offers at the end of the exclusive inventory pass.
+    # None means no authoritative inventory was completed, not an empty cache.
+    surviving_entries: tuple[EntryKey, ...] | None = None
 
 
 @dataclass(frozen=True)
@@ -2508,14 +2511,17 @@ def root_files(root: Path) -> tuple[Path, ...]:
                 else ()
             )
             chunk_sizes: dict[Path, int] = {}
+            chunk_logical_sizes: dict[str, int] = {}
             canonical_chunks: dict[str, Path] = {}
             for path in chunk_paths:
                 try:
-                    chunk_sizes[path] = _allocated_bytes(path.stat())
+                    metadata = path.stat()
+                    chunk_sizes[path] = _allocated_bytes(metadata)
                 except FileNotFoundError:
                     continue
                 if path.suffix == ".spcc" and _DIGEST.fullmatch(path.stem):
                     canonical_chunks[path.stem] = path
+                    chunk_logical_sizes[path.stem] = metadata.st_size
 
             references: Counter[str] = Counter()
             segment_references: Counter[tuple[str, str]] = Counter()
@@ -2746,6 +2752,21 @@ def select(entry: _CapacityEntry) -> None:
             aliases_removed = sum(
                 entry.key.root_kind == "prefix_alias" for entry in removed
             )
+            remaining_entries = [entry for entry in entries if entry.path not in removed_paths]
+            # Exact roots shadow aliases even when corrupt or when an unlink
+            # failed. Match lookup's exact-first policy without re-reading roots.
+            exact_keys = {
+                (entry.key.storage_key, entry.key.context_digest)
+                for entry in remaining_entries if entry.key.root_kind == "manifest"
+            }
+            surviving_entries = tuple(
+                entry.key for entry in remaining_entries
+                if entry.valid
+                and (entry.key.root_kind == "manifest"
+                     or (entry.key.storage_key, entry.key.context_digest) not in exact_keys)
+                and all(chunk_logical_sizes.get(digest) == size
+                        for digest, size in entry.chunks)
+            )
             return MaintenanceReport(
                 bytes_before=bytes_before,
                 bytes_after=bytes_after,
@@ -2762,6 +2783,7 @@ def select(entry: _CapacityEntry) -> None:
                 aliases_evicted=aliases_removed,
                 segments_deleted=segments_deleted,
                 orphan_segments_deleted=orphan_segments_deleted,
+                surviving_entries=surviving_entries,
             )
         finally:
             self._maintenance_not_before = (
diff --git a/sparkcache/spark_context_cache_connector.py b/sparkcache/spark_context_cache_connector.py
index 126aaf0..7e92caa 100644
--- a/sparkcache/spark_context_cache_connector.py
+++ b/sparkcache/spark_context_cache_connector.py
@@ -3350,6 +3350,7 @@ def _perform_capacity_maintenance_locked(
         """Keep scans and survivor reconciliation within the maintenance activity gauge."""
         try:
             with self._store_cv:
+                held_snapshot = (self._held, self._held.revision, set(self._held))
                 protected = tuple(
                     key
                     for result_digest, base in getattr(self, "_publication_base_pins", {}).items()
@@ -3413,53 +3414,17 @@ def _perform_capacity_maintenance_locked(
         self.counters["prefix_alias_segments_deleted"] += int(
             getattr(report, "segments_deleted", 0)
         )
-        if report.evicted_entries:
+        if report.surviving_entries is None:
+            # A report without an inventory cannot authorize withdrawals.
+            self._reconcile_held_capacity()
+        elif held_snapshot[2]:
             identity = self._identity(self._worker_rank())
-            withdrawn = set()
-            candidates: dict[str, set[str]] = {}
-            for entry in report.evicted_entries:
-                if entry.storage_key != identity.storage_key:
-                    continue
-                candidates.setdefault(entry.context_digest, set()).add(
-                    getattr(entry, "root_kind", "manifest")
-                )
-            for digest, evicted_roots in candidates.items():
-                exact_exists = (
-                    "manifest" not in evicted_roots
-                    and (
-                        Path(self._root)
-                        / "manifests"
-                        / identity.storage_key
-                        / f"{digest}.json"
-                    ).exists()
-                )
-                alias_exists = (
-                    self._storage_mode == "per_token_rows"
-                    and "prefix_alias" not in evicted_roots
-                    and (
-                        Path(self._root)
-                        / "prefix-aliases"
-                        / identity.storage_key
-                        / f"{digest}.json"
-                    ).exists()
-                )
-                if not exact_exists and not alias_exists:
-                    withdrawn.add(digest)
-                    continue
-                lookup, _is_alias = self._lookup_reusable(
-                    identity,
-                    digest,
-                    verify_chunks=False,
-                    verify_chunk_metadata=True,
-                )
-                if not lookup.is_hit:
-                    withdrawn.add(digest)
-            with self._store_cv:
-                self._held.difference_update(withdrawn)
-        # One digest can name both an exact manifest and its source-boundary
-        # alias. The targeted checks above retain the offer when either root
-        # remains; this complete pass also catches entries removed as debris.
-        self._reconcile_held_capacity()
+            surviving = {
+                entry.context_digest for entry in report.surviving_entries
+                if entry.storage_key == identity.storage_key
+                and (entry.root_kind == "manifest" or self._storage_mode == "per_token_rows")
+            }
+            self._apply_held_capacity_survivors(held_snapshot, surviving)
         if (not report.capacity_satisfied or report.work_pending) and wake_worker_on_unsatisfied:
             self._capacity_wakeup.set()
         if force or report.bytes_reclaimed:
@@ -3772,15 +3737,32 @@ def _finalize_streaming_capacity_commits(
                 self.counters["streaming_store_evicted"] += len(evicted)
             return True
 
+    def _apply_held_capacity_survivors(
+        self,
+        snapshot: tuple[HeldInventory, int, set[str]],
+        surviving: set[str],
+    ) -> None:
+        inventory, revision, held = snapshot
+        with self._store_cv:
+            if self._held is not inventory or self._held.revision != revision:
+                # A digest can be withdrawn and republished during a probe.
+                # An earlier inventory cannot revoke that publication; a
+                # stable pass reconciles it, and restores still verify bytes.
+                self.counters["capacity_stale_inventory_snapshots"] = (
+                    self.counters.get("capacity_stale_inventory_snapshots", 0) + 1
+                )
+                return
+            self._held.difference_update(held - surviving)
+
     def _reconcile_held_capacity(self) -> None:
-        if not self._held:
+        with self._store_cv:
+            snapshot = (self._held, self._held.revision, set(self._held))
+        if not snapshot[2]:
             return
         rank = self._worker_rank()
         identity = self._identity(rank)
-        with self._store_cv:
-            held = set(self._held)
         surviving = set()
-        for digest in held:
+        for digest in snapshot[2]:
             lookup, _is_alias = self._lookup_reusable(
                 identity,
                 digest,
@@ -3789,8 +3771,7 @@ def _reconcile_held_capacity(self) -> None:
             )
             if lookup.is_hit:
                 surviving.add(digest)
-        with self._store_cv:
-            self._held.intersection_update(surviving)
+        self._apply_held_capacity_survivors(snapshot, surviving)
 
     def _ensure_capacity_thread(self) -> None:
         if self._capacity_thread is not None:
@@ -6551,6 +6532,7 @@ def get_kv_connector_stats(self):
                 ],
                 maintenance_retries=self.counters["capacity_retries"],
                 maintenance_deletion_attempts=self.counters.get("capacity_deletion_attempts", 0),
+                maintenance_stale_inventory_snapshots=self.counters.get("capacity_stale_inventory_snapshots", 0),
                 maintenance_budget_exhausted=self.counters.get("capacity_budget_exhausted", 0),
                 maintenance_skipped_cooldown=self.counters.get("capacity_skipped_cooldown", 0),
             )
diff --git a/sparkcache/test_maintenance_survivors.py b/sparkcache/test_maintenance_survivors.py
new file mode 100644
index 0000000..76cba39
--- /dev/null
+++ b/sparkcache/test_maintenance_survivors.py
@@ -0,0 +1,186 @@
+"""A maintenance inventory qualifies offers without a second filesystem walk."""
+
+from collections import Counter
+from pathlib import Path
+
+import pytest
+
+from sparkcache.persistent_context_cache.cache_manifest import (
+    CapacityPolicy, EntryKey, LookupResult, ManifestStore,
+)
+from sparkcache.persistent_context_cache.test_cache_manifest import _chunk, _identity
+from sparkcache.persistent_context_cache.test_prefix_aliases import _publish_source
+from sparkcache.test_spark_context_cache_connector import _make_connector
+
+
+POLICY = CapacityPolicy(max_bytes=10**9, low_watermark_bytes=10**9)
+
+
+@pytest.fixture
+def connector(tmp_path):
+    connector = _make_connector(tmp_path, 0)
+    identity = _identity()
+    connector._identity = lambda rank: identity
+    connector._capacity_policy = POLICY
+    yield connector
+    connector.shutdown()
+
+
+def populate(connector, roots=12, chunks=8):
+    identity = connector._identity(0)
+    payloads = tuple(_chunk(index * 256, (index + 1) * 256) for index in range(chunks))
+    digests = {f"{index + 1:064x}" for index in range(roots)}
+    for digest in digests:
+        connector._store.commit(identity=identity, context_digest=digest, chunks=payloads)
+    connector._held.update(digests)
+    return digests
+
+
+def test_normal_maintenance_reads_each_root_once_and_stats_unique_chunks(connector, monkeypatch):
+    digests = populate(connector)
+    counts = Counter()
+    read_bytes, stat = Path.read_bytes, Path.stat
+
+    def read(path):
+        if path.suffix == ".json":
+            counts["roots"] += 1
+        return read_bytes(path)
+
+    def metadata(path, *args, **kwargs):
+        if path.parent.name == "chunks" and path.suffix == ".spcc":
+            counts["chunks"] += 1
+        return stat(path, *args, **kwargs)
+
+    monkeypatch.setattr(Path, "read_bytes", read)
+    monkeypatch.setattr(Path, "stat", metadata)
+    monkeypatch.setattr(connector, "_lookup_reusable",
+                        lambda *args, **kwargs: pytest.fail("inventory was probed twice"))
+    report = connector._maintain_capacity(force=True)
+    assert report.surviving_entries is not None
+    assert set(connector._held) == digests
+    assert counts["roots"] == 12
+    # Enumeration tests file type and obtains allocated/logical size once.
+    assert counts["chunks"] <= 2 * 8
+
+
+@pytest.mark.parametrize("damage", ["missing", "truncated", "same_size"])
+def test_shared_payload_metadata_qualifies_every_offer(connector, damage):
+    digests = populate(connector, roots=3, chunks=1)
+    chunk = next((Path(connector._root) / "chunks").glob("*.spcc"))
+    encoded = chunk.read_bytes()
+    if damage == "missing":
+        chunk.unlink()
+    elif damage == "truncated":
+        chunk.write_bytes(encoded[:-1])
+    else:
+        chunk.write_bytes(encoded[:-1] + bytes([encoded[-1] ^ 1]))
+    connector._maintain_capacity(force=True)
+    assert set(connector._held) == (digests if damage == "same_size" else set())
+    # Metadata qualification never substitutes for restore integrity checks.
+    for digest in digests:
+        assert not connector._store.lookup(connector._identity(0), digest).is_hit
+
+
+def test_corrupt_exact_root_shadows_valid_alias_until_removed(tmp_path, monkeypatch):
+    store, identity, _, digest, _ = _publish_source(tmp_path, chunk_count=4,
+                                                 prefix_tokens=(1024,))
+    exact = store._manifest_path(identity, digest)
+    exact.write_bytes(b"{}")
+    unlink = Path.unlink
+
+    def refuse_exact(path, *args, **kwargs):
+        if path == exact:
+            raise PermissionError("root retained")
+        return unlink(path, *args, **kwargs)
+
+    monkeypatch.setattr(Path, "unlink", refuse_exact)
+    report = store.maintain(POLICY)
+    assert report.surviving_entries == ()
+    assert not store.lookup(identity, digest, verify_chunks=False,
+                            verify_chunk_metadata=True, storage_mode="per_token_rows").is_hit
+    monkeypatch.setattr(Path, "unlink", unlink)
+    report = store.maintain(POLICY)
+    assert EntryKey(identity.storage_key, digest, "prefix_alias") in report.surviving_entries
+    assert store.lookup(identity, digest, storage_mode="per_token_rows").is_hit
+
+
+def test_alias_offers_are_excluded_from_block_page_connector(connector, tmp_path):
+    store, identity, _, digest, _ = _publish_source(tmp_path, chunk_count=4,
+                                                 prefix_tokens=(1024,))
+    store._manifest_path(identity, digest).unlink()
+    connector._identity = lambda rank: identity
+    connector._held.add(digest)
+    connector._storage_mode = "block_pages"
+    connector._maintain_capacity(force=True)
+    assert digest not in connector._held
+    connector._storage_mode = "per_token_rows"
+    connector._held.add(digest)
+    connector._maintain_capacity(force=True)
+    assert digest in connector._held
+
+
+@pytest.mark.parametrize("mutation", ["add", "replace_inventory", "republish_same_digest"])
+def test_inventory_mutations_are_not_withdrawn_from_a_stale_pass(connector, monkeypatch, mutation):
+    stale = "a" * 64
+    fresh = "b" * 64
+    connector._held.add(stale)
+    maintain = connector._store.maintain
+
+    def race(*args, **kwargs):
+        report = maintain(*args, **kwargs)
+        with connector._store_cv:
+            if mutation == "add":
+                connector._held.add(fresh)
+            elif mutation == "replace_inventory":
+                connector._held = {fresh}
+            else:
+                connector._held.remove(stale)
+                connector._held.add(stale)
+        return report
+
+    monkeypatch.setattr(connector._store, "maintain", race)
+    connector._maintain_capacity(force=True)
+    assert fresh in connector._held if mutation != "republish_same_digest" else stale in connector._held
+    assert connector.counters["capacity_stale_inventory_snapshots"] == 1
+    monkeypatch.setattr(connector._store, "maintain", maintain)
+    connector._maintain_capacity(force=True)
+    assert not connector._held
+
+
+def test_fallback_probe_preserves_same_digest_republication(connector, monkeypatch):
+    digest = "a" * 64
+    connector._held.add(digest)
+
+    def changed(*args, **kwargs):
+        with connector._store_cv:
+            connector._held.remove(digest)
+            connector._held.add(digest)
+        return LookupResult(False, "absent"), False
+
+    monkeypatch.setattr(connector, "_lookup_reusable", changed)
+    connector._reconcile_held_capacity()
+    assert digest in connector._held
+    assert connector.counters["capacity_stale_inventory_snapshots"] == 1
+
+
+def test_other_store_publications_are_observed_on_each_inventory(connector):
+    root = Path(connector._root)
+    assert connector._maintain_capacity(force=True).surviving_entries == ()
+    other = ManifestStore(root)
+    digest = "a" * 64
+    identity = connector._identity(0)
+    other.commit(identity=identity, context_digest=digest, chunks=[_chunk()])
+    connector._held.add(digest)
+    report = connector._maintain_capacity(force=True)
+    assert report.surviving_entries == (EntryKey(identity.storage_key, digest),)
+    assert digest in connector._held
+    other._manifest_path(identity, digest).write_bytes(b"{}")
+    connector._maintain_capacity(force=True)
+    assert digest not in connector._held
+
+
+def test_empty_offer_inventory_does_not_require_initialized_model_identity(connector, monkeypatch):
+    monkeypatch.setattr(connector, "_identity",
+                        lambda rank: pytest.fail("model identity is not initialized"))
+    assert connector._maintain_capacity(force=True).surviving_entries == ()
+    connector._reconcile_held_capacity()
diff --git a/sparkcache/test_spark_context_cache_connector.py b/sparkcache/test_spark_context_cache_connector.py
index 6aaade9..6e91949 100644
--- a/sparkcache/test_spark_context_cache_connector.py
+++ b/sparkcache/test_spark_context_cache_connector.py
@@ -3548,6 +3548,7 @@ def test_eviction_withdraws_held_digest_and_updates_capacity_metrics(self) -> No
                     manifests_evicted=1,
                     chunks_deleted=2,
                     evicted_entries=(EntryKey(storage_key, removed),),
+                    surviving_entries=(EntryKey(storage_key, survivor),),
                 )
             )
             connector._reconcile_held_capacity = mock.Mock()
diff --git a/tools/benchmark_maintenance_inventory.py b/tools/benchmark_maintenance_inventory.py
new file mode 100644
index 0000000..a1f8909
--- /dev/null
+++ b/tools/benchmark_maintenance_inventory.py
@@ -0,0 +1,108 @@
+"""Measure complete maintenance and offer reconciliation using GPU-free fixtures.
+
+Run from the checkout being measured. Positional arguments specify root/chunk
+counts for BENCH_KIND=rows (default), or branch/extension counts for pages.
+BENCH_FILESYSTEM records the independently verified filesystem type. Test
+fixtures supply connector API stubs; CUDA and model execution are not used.
+"""
+from collections import Counter
+import gc
+import os
+import platform
+import hashlib
+import json
+from pathlib import Path
+import statistics
+import subprocess
+import sys
+import tempfile
+import time
+from unittest.mock import patch
+
+sys.path.insert(0, str(Path.cwd()))
+from sparkcache.persistent_context_cache.cache_manifest import CapacityPolicy
+from sparkcache.persistent_context_cache.test_cache_manifest import _chunk, _identity
+from sparkcache.test_spark_context_cache_connector import _make_connector
+
+roots, chunks = map(int, sys.argv[1:3])
+with tempfile.TemporaryDirectory(prefix="sparkcache-inventory-bench-") as directory:
+    connector = _make_connector(Path(directory), 0)
+    identity = _identity()
+    connector._identity = lambda rank: identity
+    connector._capacity_policy = CapacityPolicy(max_bytes=10**9, low_watermark_bytes=10**9)
+    kind = os.environ.get("BENCH_KIND", "rows")
+    if kind == "pages":
+        from sparkcache.persistent_context_cache.cache_manifest import CacheIdentity
+        from sparkcache.spark_context_cache_codec import context_prefix_digest
+        from sparkcache.spark_context_cache_hybrid import PageGroup, PageLayer, PageLayout, encode_page_snapshot
+        identity = CacheIdentity(target_checkpoint="1" * 64, draft_checkpoint="2" * 64,
+                                 quantization_layout="benchmark-page", rope_layout="glm53-hybrid-v1",
+                                 tp_degree=4, dcp_degree=4, chunk_tokens=256,
+                                 record_schema=("target_ckv", "logical_positions"),
+                                 publication_schema="page-tail-cow-v2")
+        connector._identity = lambda rank: identity
+        connector._storage_mode = "block_pages"
+        layout = PageLayout((PageGroup(256, (PageLayer("attention", "u8", (64,), 64),)),
+                             PageGroup(1, (PageLayer("recurrent", "u8", (32,), 32),))))
+        tokens_base = tuple(range(512))
+        salt = "shared-base-page-profile"
+        base_digest = context_prefix_digest(tokens_base, salt, token_count=512)
+        connector._store.commit_page_snapshot(identity=identity, context_digest=base_digest,
+            span_tokens=512, snapshot=encode_page_snapshot(layout, (2, 1),
+            {"attention": b"a" * 128, "recurrent": b"r" * 32}))
+        for branch in range(roots):
+            tokens = tokens_base + tuple(range(100000 * (branch + 1), 100000 * (branch + 1) + chunks * 256))
+            prior = base_digest
+            payload = b"a" * 128
+            for stage in range(chunks):
+                payload += bytes((stage + branch,)) * 64
+                boundary = (stage + 3) * 256
+                snapshot = encode_page_snapshot(layout, (stage + 3, 1), {
+                    "attention": payload, "recurrent": bytes((stage + 64,)) * 32})
+                connector._store.commit_page_extension(identity=identity, base_context_digest=prior,
+                    token_ids=tokens, identity_salt=salt, layout=layout,
+                    base_block_counts=(stage + 2, 1), result_block_counts=(stage + 3, 1),
+                    base_boundary_tokens=(stage + 2) * 256,
+                    result_boundary_tokens=boundary, result_snapshot=snapshot)
+                prior = context_prefix_digest(tokens, salt, token_count=boundary)
+        digests = {path.stem for path in (Path(directory) / "manifests" / identity.storage_key).glob("*.json")}
+        references = sum(len(connector._store._capacity_entry(path).chunks)
+                         for path in (Path(directory) / "manifests" / identity.storage_key).glob("*.json"))
+    else:
+        payloads = tuple(_chunk(i * 256, (i + 1) * 256) for i in range(chunks))
+        digests = {f"{i + 1:064x}" for i in range(roots)}
+        for digest in digests:
+            connector._store.commit(identity=identity, context_digest=digest, chunks=payloads)
+        references = roots * chunks
+    unique_chunks = len(list((Path(directory) / "chunks").glob("*.spcc")))
+    connector._held.update(digests)
+    counts = Counter()
+    read_bytes, stat = Path.read_bytes, Path.stat
+
+    def read(path):
+        if path.suffix == ".json":
+            counts["root_reads"] += 1
+        return read_bytes(path)
+
+    def metadata(path, *args, **kwargs):
+        if path.parent.name == "chunks" and path.suffix == ".spcc":
+            counts["chunk_stats"] += 1
+        return stat(path, *args, **kwargs)
+
+    with patch.object(Path, "read_bytes", read), patch.object(Path, "stat", metadata):
+        connector._maintain_capacity(force=True)
+    elapsed = []
+    for _ in range(7):
+        gc.collect()
+        start = time.perf_counter()
+        report = connector._maintain_capacity(force=True)
+        elapsed.append((time.perf_counter() - start) * 1000)
+        assert set(connector._held) == digests
+        assert report.manifests_evicted == report.chunks_deleted == 0
+    connector.shutdown()
+    print(json.dumps({"revision": subprocess.check_output(["git", "rev-parse", "HEAD"], text=True).strip(),
+                      "kind": kind, "roots": len(digests), "unique_chunks": unique_chunks, "references": references,
+                      "platform": platform.platform(), "filesystem": os.environ.get("BENCH_FILESYSTEM", "unspecified"),
+                      "manifest_source_sha256": hashlib.sha256(Path('sparkcache/persistent_context_cache/cache_manifest.py').read_bytes().replace(b'\r\n', b'\n')).hexdigest(),
+                      "connector_source_sha256": hashlib.sha256(Path('sparkcache/spark_context_cache_connector.py').read_bytes().replace(b'\r\n', b'\n')).hexdigest(),
+                      "io_counts": counts, "elapsed_ms": elapsed, "median_ms": statistics.median(elapsed)}))

From ce0dca2cfa45ee0aee2290ff94f9ee7126820a38 Mon Sep 17 00:00:00 2001
From: FujitsuPolycom <87842395+FujitsuPolycom@users.noreply.github.com>
Date: Sun, 6 Sep 2026 18:04:43 -0500
Subject: [PATCH 6/7] Document source instrumentation and maintenance
 qualification limits

Separate source-build capabilities from the immutable published image, describe cumulative accepted-prompt accounting, and state the remaining full-scan and workload-validation limits. No runtime behavior or cache identity changes.
---
 deploy/glm53_flash/MTP3_STATUS.md | 28 +++++++++++++++++++++-------
 1 file changed, 21 insertions(+), 7 deletions(-)

diff --git a/deploy/glm53_flash/MTP3_STATUS.md b/deploy/glm53_flash/MTP3_STATUS.md
index 15840ad..1427ac6 100644
--- a/deploy/glm53_flash/MTP3_STATUS.md
+++ b/deploy/glm53_flash/MTP3_STATUS.md
@@ -29,21 +29,35 @@ identified above. It requires a rebuilt image and startup validation.
 
 ## What remains in the performance issues
 
+Source builds include request-attribution accounting, same-pass maintenance
+inventory reuse, and optional deletion pacing. The immutable image identified
+above does not contain these additions. Attribution also requires the matching
+instrumented SparkRing scheduler; a package update alone cannot supply its events.
+
 | Issue | Implemented | Remaining work |
 |---|---|---|
-| [#60: sustained publication and eviction slowdown](https://github.com/FujitsuPolycom/sparkcache/issues/60) | Fewer repeated restore reads and hashes, bounded restore arenas, tiled native placement, publication-dependency protection, publication-backlog gauges, optional periodic full captures, and capacity guidance. | Maintenance still scans and sorts the inventory and evicts toward the low watermark; it has no per-pass time or entry budget. Reproduce the reported slowdown with a near-full 40 GiB store and matched before/after probes. |
-| [#61: growing conversations lose local prefix reuse](https://github.com/FujitsuPolycom/sparkcache/issues/61) | Runtime GPU-lease accounting, preference for longer local prefixes, recurrent checkpoint retention, and opt-in restore/lease traces. | Exact per-request local-hit, verified-restore, and recompute token counters are not implemented. Validate local retention on the reported long-context, multi-turn workload with occasional images. |
+| [#60: sustained publication and eviction slowdown](https://github.com/FujitsuPolycom/sparkcache/issues/60) | Reduced restore work, publication-dependency protection, backlog gauges, capacity guidance, same-pass inventory reuse, and opt-in deletion-attempt limits and cooldown. | Full inventory scans and individual filesystem operations still have no hard time bound. Validate the combined implementation against near-full 40 GiB traffic and matched before/after probes. |
+| [#61: growing conversations lose local prefix reuse](https://github.com/FujitsuPolycom/sparkcache/issues/61) | Runtime lease accounting, local-prefix preference, checkpoint retention, and opt-in request attribution for accepted target execution. | Deploy the matching scheduler instrumentation and validate attribution and local retention on the reported long-context conversations with occasional images. |
 
 One saver admission per rank bounds concurrent optional work; it does not bound
-the duration of an inventory scan or eviction pass. See
+the duration of an inventory scan. Deletion pacing is disabled by default.
+Smaller deletion budgets can repeat scans and increase total work. See
 [capacity and cleanup](../../sparkcache/README.md#capacity-and-cleanup).
 Backlog reports distinguish pending publications, their oldest age, and active
 maintenance, but reports may not refresh while the connector is idle.
 
-Reuse traces distinguish restore offers, verified worker completion, and GPU
-lease attachment. Their scheduler prefix-token field is block-aligned input,
-not an exact local hash-hit measurement. API `cached_tokens` alone is not enough
-to attribute reuse to GPU retention rather than persistent restoration.
+Matching traces distinguish restore offers, worker completion, and GPU lease
+attachment. Their scheduler prefix-token input and API `cached_tokens` alone
+cannot establish exact attribution.
+
+The [request-attribution record](../../sparkcache/README.md#request-reuse-attribution)
+uses authoritative scheduler events to distinguish consumed local prefixes,
+verified external reuse, and accepted prompt computation. Counts accumulate
+across preemption attempts; incomplete observations are labeled explicitly.
+
+The [maintenance inventory benchmark](../../docs/maintenance-inventory-validation.md)
+measures reduced metadata work on CPU fixtures. It is not a DGX4 serving result
+or confirmation that the original slowdown is resolved.
 
 Both issues should remain open until their remaining implementation questions
 and workload-specific results are recorded explicitly.

From b5aca7cd3d3f7e7a14636bf6e5fa1f50a9650168 Mon Sep 17 00:00:00 2001
From: FujitsuPolycom <87842395+FujitsuPolycom@users.noreply.github.com>
Date: Sun, 6 Sep 2026 18:05:33 -0500
Subject: [PATCH 7/7] Bind maintenance measurements to matching source
 revisions

Resolve the measured baseline from its recorded source hashes and retain the combined-source repeat with its own timings and hashes. This evidence records CPU maintenance behavior only; it does not claim serving qualification.
---
 docs/maintenance-inventory-validation.json | 89 +++++++++++++++++++---
 docs/maintenance-inventory-validation.md   |  7 +-
 2 files changed, 85 insertions(+), 11 deletions(-)

diff --git a/docs/maintenance-inventory-validation.json b/docs/maintenance-inventory-validation.json
index 8266459..16e5a3d 100644
--- a/docs/maintenance-inventory-validation.json
+++ b/docs/maintenance-inventory-validation.json
@@ -8,8 +8,17 @@
     "python_execution": "native Windows",
     "paired_method": "Separate sequential processes constructed equivalent temporary caches and invoked SparkContextCacheConnector._maintain_capacity(force=True).",
     "timing": "Median of seven complete maintenance/reconciliation passes after an untimed instrumented pass; garbage collection precedes each timer.",
-    "excluded": ["fixture construction", "publication", "Python imports", "I/O counting instrumentation"],
-    "included": ["exclusive cache lock", "complete reference inventory", "offer reconciliation"],
+    "excluded": [
+      "fixture construction",
+      "publication",
+      "Python imports",
+      "I/O counting instrumentation"
+    ],
+    "included": [
+      "exclusive cache lock",
+      "complete reference inventory",
+      "offer reconciliation"
+    ],
     "eviction": "disabled by a 1000000000-byte capacity ceiling",
     "gpu_execution": false,
     "serving_or_nvme_qualification": false,
@@ -17,9 +26,10 @@
   },
   "sources": {
     "baseline": {
-      "revision": "78ca0bdc5d63629ef6ee6239546921898956fa2e",
+      "revision": "7fc9f85509624a0f9c5bc6bd7c5f70f1a43d389e",
       "cache_manifest_sha256_lf": "88caa5b104407a212b4344b82ab914d6c5c7e72d021976ba6d1c44d13f7f2f2e",
-      "connector_sha256_lf": "082f2820f45cdf951629d045b5d316461f86372fae70425be9b5903825395a97"
+      "connector_sha256_lf": "082f2820f45cdf951629d045b5d316461f86372fae70425be9b5903825395a97",
+      "revision_matches_recorded_source_hashes": true
     },
     "survivor_inventory": {
       "mechanism": "MaintenanceReport.surviving_entries supplies metadata-qualified offers from the exclusive inventory pass.",
@@ -39,13 +49,29 @@
     "baseline": {
       "root_reads": 514,
       "chunk_stats": 4995,
-      "elapsed_ms": [185.0543999898946, 183.6131000018213, 173.09400001249742, 178.17310000828002, 174.59690000396222, 174.0127999946708, 175.22859999735374],
+      "elapsed_ms": [
+        185.0543999898946,
+        183.6131000018213,
+        173.09400001249742,
+        178.17310000828002,
+        174.59690000396222,
+        174.0127999946708,
+        175.22859999735374
+      ],
       "median_ms": 175.22859999735374
     },
     "survivor_inventory": {
       "root_reads": 257,
       "chunk_stats": 514,
-      "elapsed_ms": [72.49789999332279, 70.9155000076862, 72.2200000018347, 70.33850000880193, 76.28729999123607, 89.75840000493918, 72.21110000682529],
+      "elapsed_ms": [
+        72.49789999332279,
+        70.9155000076862,
+        72.2200000018347,
+        70.33850000880193,
+        76.28729999123607,
+        89.75840000493918,
+        72.21110000682529
+      ],
       "median_ms": 72.2200000018347
     },
     "conclusion": "Complete pass latency decreased 58.8 percent in this CPU fixture. Each root was read once; shared-chunk metadata probes scaled with unique files rather than root references. This does not establish a serving-throughput improvement."
@@ -55,14 +81,57 @@
     "unique_chunks": 32,
     "chunk_references": 2048,
     "baseline": {
-      "root_reads": 128, "chunk_stats": 2112,
-      "elapsed_ms": [37.09349999553524, 37.26160000951495, 37.37550000369083, 37.524400002439506, 37.23980000359006, 41.24779999256134, 52.597900008549914],
+      "root_reads": 128,
+      "chunk_stats": 2112,
+      "elapsed_ms": [
+        37.09349999553524,
+        37.26160000951495,
+        37.37550000369083,
+        37.524400002439506,
+        37.23980000359006,
+        41.24779999256134,
+        52.597900008549914
+      ],
       "median_ms": 37.37550000369083
     },
     "survivor_inventory": {
-      "root_reads": 64, "chunk_stats": 64,
-      "elapsed_ms": [10.044100010418333, 10.23929999792017, 10.25409999419935, 10.973099997499958, 10.138300000107847, 9.93929999822285, 9.973099993658252],
+      "root_reads": 64,
+      "chunk_stats": 64,
+      "elapsed_ms": [
+        10.044100010418333,
+        10.23929999792017,
+        10.25409999419935,
+        10.973099997499958,
+        10.138300000107847,
+        9.93929999822285,
+        9.973099993658252
+      ],
       "median_ms": 10.138300000107847
     }
+  },
+  "combined_source_repeat": {
+    "revision": "11658f0a3b9f155af1d95ff78a807c9b374f6093",
+    "kind": "pages",
+    "roots": 257,
+    "unique_chunks": 257,
+    "references": 4481,
+    "platform": "Windows-11-10.0.26200-SP0",
+    "filesystem": "NTFS",
+    "manifest_source_sha256": "f2cc0bc7570b59ed068ad3273b234c0f8c9bd945a6a68dca0014012a118d1400",
+    "connector_source_sha256": "f92b1fcaf3385e5d4487b246f06c1de3fb02e8f05b67c055ab760d91ccd86eb4",
+    "io_counts": {
+      "root_reads": 257,
+      "chunk_stats": 514
+    },
+    "elapsed_ms": [
+      76.01800000702497,
+      72.13889999547973,
+      73.28540000889916,
+      82.84220000496134,
+      89.14029999868944,
+      79.15170000342187,
+      78.41010000265669
+    ],
+    "median_ms": 78.41010000265669
   }
 }
diff --git a/docs/maintenance-inventory-validation.md b/docs/maintenance-inventory-validation.md
index 5191be6..7f6ee9a 100644
--- a/docs/maintenance-inventory-validation.md
+++ b/docs/maintenance-inventory-validation.md
@@ -8,13 +8,18 @@ rereading roots or probing shared payload files once per reference.
 
 The [CPU benchmark record](maintenance-inventory-validation.json) measures the
 complete maintenance and reconciliation call on Windows 11 with NTFS. It
-compares equivalent fixtures against source revision `78ca0bd`.
+compares equivalent fixtures against source revision `7fc9f85509624a0f9c5bc6bd7c5f70f1a43d389e`.
 
 | Fixture | Baseline median | Inventory median | Root reads | Chunk metadata probes |
 |---|---:|---:|---:|---:|
 | Eight branches, 32 flat page extensions each | 175.23 ms | 72.22 ms | 514 to 257 | 4,995 to 514 |
 | 64 token-row roots sharing 32 chunks | 37.38 ms | 10.14 ms | 128 to 64 | 2,112 to 64 |
 
+A repeat with combined source `11658f0a3b9f155af1d95ff78a807c9b374f6093`,
+including canonical string ordering, measured 78.41 ms for the flat-page fixture
+and retained the same reduced I/O counts. The JSON records that source and its
+seven observations separately.
+
 The page fixture includes attention and recurrent-state bytes. It exercises
 257 roots and 4,481 chunk references with tiny payloads, not a loaded model.
 The measurements do not qualify DGX serving throughput or NVMe behavior.