feat: add source-owned move with (deferred) G2 removal

docs/design_overview.md

@@ -249,22 +249,36 @@ kvcr.discard_hint(request_id) # request is over or its hints should be discarded
 
 ### Inventory and block identity
 
-The router maintains an eventually-consistent global prefix inventory for cache-aware request routing. The router is the routing-key authority and supplies the request's content-derived `BlockKey` values. A `BlockKey` identifies one cacheable KV unit, while a `BlockKeyList` is an ordered exact request path rather than an implicit prefix expansion. The framework normally passes these keys unchanged to the KVCR; an integration with distinct local hashes translates them at its boundary. The key space may be namespaced by model version, quantization, KV layout, tenant, or hash algorithm, and the KVCR treats each full key as opaque.
+The router maintains an eventually-consistent global prefix inventory for cache-aware request routing. Its advertised hashes may differ from the full keys used by the engine's cache. The framework supplies exact `BlockKey` values to KVCR, including any representation-specific identity such as a cache-group ID. A `BlockKey` identifies one cacheable KV unit, while a `BlockKeyList` is an ordered exact request path rather than an implicit prefix expansion. The key space may be namespaced by model version, quantization, KV layout, tenant, or hash algorithm, and KVCR treats each full key as opaque.
 
 Framework-owned memory is reported through the framework's existing router event path. If needed, `tier_enter`/`tier_exit` reporting or `inventory_snapshot()` can be added to the KVCR API later.
 
-KVCR-owned inventory events use the same `BlockKey` values as fetch and `submit_hint` in the normal case. KVCR deliberately reports only the affected keys, tier, and whether the residency was removed. KVCR delivers each event to `inventory_sink`; the framework or local indexer owns any buffering or batching for publication. Each event may contain multiple keys and is tier-specific because one block may be present in several tiers at once.
+KVCR-owned inventory events use the same full `BlockKey` values as fetch. Request-scoped hints instead carry hashes interpreted by the configured key adapter. KVCR deliberately reports only the affected keys, tier, and whether the residency was removed. KVCR delivers each event to `inventory_sink`; the framework or local indexer owns any buffering or batching for publication. Each event may contain multiple keys and is tier-specific because one block may be present in several tiers at once.
 
 ### Request-scoped hints
 
 For data shared among engines on the same node, the router can use one KVCR as the source in hints sent to the other local KVCRs. A destination may retain a local copy when replication is worthwhile. Otherwise, the router includes `no_retain` in `submit_hint`. The destination still satisfies any committed framework request; `no_retain` only advises it not to keep an additional KVCR-owned copy after the framework no longer needs it.
 
-KVCR-to-KVCR movement is destination-initiated. A router hint gives the destination the source node and KVCR control endpoint. `mode=copy` retains the source copy, while `mode=move` permits source eviction only after successful completion. Timeout or cancellation keeps the source copy. When the hint has no source control endpoint, the destination uses its local storage information or checks object storage when enabled. The router does not need object-store inventory for locality decisions, but may track object presence to issue source hints.
+KVCR-to-KVCR movement is destination-initiated. A router hint gives the destination the source node and KVCR control endpoint. `mode=copy` retains the source copy, while `mode=move` lets the source remove its G2 copy after its native transfer succeeds. When the hint has no source control endpoint, the destination uses its local storage information or checks object storage when enabled. The router does not need object-store inventory for locality decisions, but may track object presence to issue source hints.
 
 `submit_hint` may establish the peer connection, but does not start data movement or remote pinning yet. Each hint in an envelope is parsed and handled independently; proactive fetching, staging, or pinning may be added later as new hint types. `discard_hint` normally reports that the relevant request is over, and also permits an integration to discard the request-scoped hint early. A route-time `submit_hint` may carry a whole `BlockKeyList` from one source. In the future, if necessary, multi-source assembly can be added by splitting the list into multiple hinted operations.
 
 The router may choose not to send a hint based on its internal cost model. Router hints may later be extended to request proactive copy or move for cache rebalancing, which may require periodic utilization reports from KVCR. The router also considers sending hints for matching TP layouts, meaning prefill-to-prefill, decode-to-decode, and aggregated deployments, which cover many use cases. A TP-independent host representation further removes that constraint.
 
+In the current implementation, `kv.fetch` requires a source control endpoint and carries advertised block hashes. The framework supplies complete `BlockKey` values with the same request ID; the key adapter matches those keys against the hint hashes. There is no hash-to-key index for autonomous operations on unrelated hashes, and `no_retain` and source-less hints are not implemented.
+
+`mode="copy"` is the default. With usable `mode="move"` advice, `fetch` requests source G2 removal for keys whose hashes match the hint, including destination cache hits. Move hints always require hash matching, even when opportunistic querying is enabled. Other keys retain ordinary local fetch behavior, and failed or conflicting advice does not prevent local cache hits. `deliver` always copies; a request using both calls can therefore move fetched keys while copying delivered keys. `discard_hint` clears advice without cancelling already-started operations.
+
+### Move and source invalidation
+
+The move handler captures the source advice and matching keys when `fetch` is admitted. The transfer request carries a move flag. Source keys with incomplete G2 copies or unfinished deposits fail individually before withdrawal; their deposits can still finish. The source admits moves on its owner thread, where the affected keys stop accepting new reads and deposits while existing operations retain their claims; the write itself still runs on the progress thread. Successful native transfer leaves removal pending before the source transfer claim is released; failed transfer reopens admission and keeps the copy. There is no destination completion confirmation or wait for source cleanup.
+
+A destination reuses complete copies it already holds and immediately sends the source one `invalidate` message naming them. For partly cached keys, unfinished destination writes settle before the source transfers the remaining pieces. A move joining an existing copy waits for that copy, then handles the complete key as a cache hit. Results contain ordinary fetch release handles protecting the destination copies.
+
+The destination uses `abandon_timeout_ms` as its move completion allowance; the source retains its normal transfer timeout. This tolerates delayed completion processing, but a destination that exceeds the longer allowance can still fail after the source has removed its copy. Existing cancellation and buffer-quarantine rules continue to protect memory until native access stops.
+
+Removal affects only KVCR-owned G2 copies in the requested complete layout, after earlier operations and readers drain. `deposit(no_evict=True)` protection is honored; framework-owned memory and G3 copies are retained. Actual source removal and newly complete destination copies publish the corresponding `inventory_sink` events. Removal remains best effort, with no separate public outcome. Control messages assume trusted peers, and no removal is requested when the hint names this node as its own source.
+
 ### Peer control and liveness
 
 The router is never on the data path: KVCR instances execute transfers peer-to-peer via NIXL. Source endpoints remain request-scoped rather than becoming part of global block state, so this coordination adds no KVCR-side peer inventory.
@@ -313,7 +327,7 @@ At initialization, an integration selects one built-in policy or supplies an ext
 
 Eviction scores are finite floats. LRU favors evicting later sequence keys at comparable recency. Custom policies may use ordering position or ignore it.
 
-Policy decides whether to stage, retain, move, or drop data within KVCR-owned DRAM and downstream storage. Fetch and `no_evict` deposit claims are hard mechanism constraints: while they exist, policy cannot evict the entry. `no_retain` is an advisory router retention hint. If it conflicts with a framework `no_evict` claim, `no_evict` wins until the framework calls `release` with the returned handle; policy may then honor or override `no_retain`.
+Policy decides whether to stage, retain, move, or drop data within KVCR-owned DRAM and downstream storage. Fetch and `no_evict` deposit claims prevent eviction while held. Move uses the source invalidation rules above rather than a policy eviction decision. `no_retain` is not currently supported.
 
 The policy methods are invoked as follows:
 

src/kvcr/api.py

@@ -187,7 +187,15 @@ def fetch(
         expected_layout: list[str] | None = None,
         hints: object | None = None,
     ) -> OpHandle:
-        """Asynchronously fetch blocks into KVCR-managed storage."""
+        """Asynchronously fetch blocks into KVCR storage and return release claims.
+
+        A usable kv.fetch hint with mode="move" requests best-effort source G2
+        removal for matching successful keys, including destination cache hits.
+        Other keys retain ordinary fetch behavior. Failed advice does not
+        prevent local cache hits. Completion reports destination readiness
+        without waiting for source cleanup, which respects no_evict protection
+        and leaves G3 and framework memory intact.
+        """
         return self._core.fetch(keys, request_id, expected_layout, hints)
 
     def release(

src/kvcr/backends/local_dram.py

@@ -38,6 +38,7 @@
 
 if TYPE_CHECKING:
     from ..runtime.core import _KVCRCore
+    from .moves import _PendingMove
 
 logger = logging.getLogger(__name__)
 
@@ -59,6 +60,8 @@ class _PendingResidencyOp(_Op):
     key_layout: tuple[str, ...] | None = None
     fill_sources: Mapping[BlockKey, CacheTier] = field(default_factory=dict)
     claimed_keys: set[BlockKey] = field(default_factory=set)
+    # Move advice captured at admission; see _PendingMove.
+    move: "_PendingMove | None" = None
 
 
 @dataclass
@@ -247,9 +250,23 @@ def deposit(
             key_layout=key_layout,
         )
         self._pending_residency_ops[op.op_id] = op
-        self._kvcr._add_block_dependencies(op, new_operation=True)
+        self._admit(op)
         self._start_deposits(op, blocks)
 
+    def _admit(self, op: _PendingResidencyOp) -> None:
+        """Decide admission once: keys pending removal fail now and never hold a
+        dependency, so work on other keys cannot delay their removal. Accepted
+        keys may finish even if removal becomes pending later.
+        """
+        rejected = self._kvcr._invalidation_handler.rejected_entries(op.keys)
+        if rejected:
+            op.results.update(rejected)
+        self._kvcr._add_block_dependencies(
+            op,
+            new_operation=True,
+            keys=op.keys - op.results.keys() if rejected else None,
+        )
+
     def _start_deposits(
         self, op: _PendingResidencyOp, keys: Collection[BlockKey]
     ) -> None:
@@ -394,6 +411,7 @@ def fetch(
         hints: object | None,
         layout: list[str],
         key_layout: tuple[str, ...] | None = None,
+        move: "_PendingMove | None" = None,
     ) -> dict[BlockKey, list[_TransferRef]]:
         ordered_keys = tuple(dict.fromkeys(keys))
         if not ordered_keys:
@@ -409,9 +427,10 @@ def fetch(
             key_layout=key_layout,
             fill_sources=sources,
             hints=hints,
+            move=move,
         )
         self._pending_residency_ops[op.op_id] = op
-        self._kvcr._add_block_dependencies(op, new_operation=True)
+        self._admit(op)
         destinations = {}
         for key in ordered_keys:
             refs = self._prepare_fetch(op, key)
@@ -423,6 +442,8 @@ def fetch(
     def _prepare_fetch(
         self, op: _PendingResidencyOp, key: BlockKey
     ) -> list[_TransferRef] | None:
+        if key in op.results:
+            return None
         layout = op.requested_layout
         record = self._kvcr._block_record(key)
         residency = record.local_dram
@@ -463,12 +484,22 @@ def _prepare_fetch(
             ready = [residency.slots[index] for index in indices]
             self._kvcr._record_access((key,))
             op.results[key] = self._pools._new_public_claim(key, residency, slots=ready)
+            if op.move is not None and key in op.move.keys:
+                op.move.keys.remove(key)
+                op.move.cached_keys.append(key)
         else:
             if key not in op.claimed_keys:
                 self._pools._acquire_claim(key, residency)
                 op.claimed_keys.add(key)
             if source is None or residency.remote_indices is not None:
                 return None
+            if (
+                op.move is not None
+                and key in op.move.keys
+                and residency.pending_deposits
+            ):
+                # Source removal requires every piece outside this write to be ready.
+                return None
             # One remote fill owns these destinations until its native terminal.
             missing = [
                 index
@@ -783,6 +814,7 @@ def _resolve_fetch_waiter(
                 {key: refs},
                 residency_op.request_id,
                 residency_op.deadline,
+                move=residency_op.move,
             )
         return key in residency_op.results
 
@@ -898,6 +930,9 @@ def _finish_delivery_copy(self, copy: _LocalCopyOp) -> None:
             self._finish_deliver_if_ready(op)
 
     def _finish_residency_if_ready(self, op: _PendingResidencyOp) -> None:
+        if op.move is not None:
+            # Complete local hits ask the source to drop its copies right away.
+            self._kvcr._move_handler.invalidate_cached(op.move)
         if (
             len(op.results) != len(op.keys)
             or self._pending_residency_ops.get(op.op_id) is not op