feat(refit): add generator peer refit strategy

docs/ARCHITECTURE.md

@@ -462,6 +462,17 @@ engine-specific parameter and CUDA-graph handling out of the transfer layer.
 | `DeleteVersionLease` | Release a generator's protection of the version shards |
 
 The final missing source slot atomically changes the version to `READY`.
+
+Peer generators reuse the existing inference P2P metadata service rather than
+creating a second Refit peer registry. A generator serving an exact version
+publishes its normal `SourceIdentity` with `revision=WeightVersion.uid`; another
+generator queries `P2pService.ListSources` with the same engine-compatible
+identity and selects a READY source for its worker rank before falling back to
+trainer shard publications. Applied generators publish their verified canonical
+staging buffers—not engine-specific packed kernel tensors—under that identity.
+An identical-rank peer pulls those buffers directly with NIXL, then uses the
+same graph-safe engine installer as a trainer update.
+
 `WeightVersion.uid` is MX's opaque identity. `version_number` is the optional
 framework-provided numeric label used for correlation; MX does not use it as an
 identity or ordering key.

modelexpress_client/python/modelexpress/load_strategy/base.py

@@ -261,24 +261,32 @@ def unpublish_metadata(ctx: LoadContext) -> None:
     Call publish_metadata() again after memory is valid to re-enter the
     P2P network.
     """
+    unpublish_metadata_for_worker(
+        worker_rank=ctx.worker_rank,
+        device_id=ctx.device_id,
+    )
+
+
+def unpublish_metadata_for_worker(*, worker_rank: int, device_id: int) -> None:
+    """Stop one worker's publication without requiring a boot-load context."""
     from ..metadata.publish import _heartbeat_threads, _worker_servers
 
-    hb = _heartbeat_threads.pop(ctx.worker_rank, None)
+    hb = _heartbeat_threads.pop(worker_rank, None)
     if hb is not None:
         try:
             hb.stop()  # also marks STALE on MX server
-            logger.info(f"[Worker {ctx.global_rank}] Heartbeat stopped")
+            logger.info(f"[Worker {worker_rank}] Heartbeat stopped")
         except Exception as e:
             logger.warning(
-                f"[Worker {ctx.global_rank}] Failed to stop heartbeat cleanly: {e}"
+                f"[Worker {worker_rank}] Failed to stop heartbeat cleanly: {e}"
             )
 
-    ws = _worker_servers.pop(ctx.device_id, None)
+    ws = _worker_servers.pop(device_id, None)
     if ws is not None:
         try:
             ws.stop()
-            logger.info(f"[Worker {ctx.global_rank}] Worker gRPC server stopped")
+            logger.info(f"[Worker {worker_rank}] Worker gRPC server stopped")
         except Exception as e:
             logger.warning(
-                f"[Worker {ctx.global_rank}] Failed to stop worker gRPC server cleanly: {e}"
+                f"[Worker {worker_rank}] Failed to stop worker gRPC server cleanly: {e}"
             )

modelexpress_client/python/modelexpress/metadata/publish.py

@@ -93,8 +93,7 @@ def publish_metadata_and_ready(
 
         host = _get_worker_host()
 
-        grpc_base = envs.MX_WORKER_GRPC_PORT
-        worker_grpc_port = grpc_base + device_id
+        worker_grpc_port = envs.MX_WORKER_GRPC_PORT + device_id
 
         grpc_server = WorkerGrpcServer(
             tensor_protos=tensor_protos,

modelexpress_client/python/modelexpress/nixl_transfer.py

@@ -773,6 +773,7 @@ def receive_from_source(
         timeout_seconds: float | None = None,
         remote_agent_name: str | None = None,
         require_exact_match: bool = False,
+        destination_tensors: dict[str, torch.Tensor] | None = None,
     ) -> tuple[int, int, float]:
         """
         Receive weights from a remote source via NIXL RDMA.
@@ -799,6 +800,8 @@ def receive_from_source(
                 derived tensors, which would otherwise leave part or all of the
                 target at dummy values while RDMA reports success. Same-family
                 transfers leave this False and tolerate subset transfers.
+            destination_tensors: Optional registered destination catalog used for
+                name matching. Defaults to the most recently registered catalog.
 
         Returns:
             Tuple of (total_bytes, total_tensors, duration)
@@ -813,6 +816,9 @@ def receive_from_source(
 
         start_time = time.perf_counter()
         self._accelerator_backend.set_device(self._device_id)
+        local_tensors = (
+            self._tensors if destination_tensors is None else destination_tensors
+        )
 
         if remote_agent_name is None:
             add_start = time.perf_counter()
@@ -833,7 +839,7 @@ def receive_from_source(
         total_bytes = 0
 
         for src_tensor in source_tensors:
-            local_tensor = self._tensors.get(src_tensor.name)
+            local_tensor = local_tensors.get(src_tensor.name)
             if local_tensor is None:
                 continue
             local_size = local_tensor.numel() * local_tensor.element_size()
@@ -866,8 +872,8 @@ def receive_from_source(
         # Name-set diff between the source manifest and the locally registered
         # tensors.
         src_names = {s.name for s in source_tensors}
-        local_only = sorted(set(self._tensors) - src_names)
-        source_only = sorted(src_names - set(self._tensors))
+        local_only = sorted(set(local_tensors) - src_names)
+        source_only = sorted(src_names - set(local_tensors))
         if local_only or source_only:
             if require_exact_match:
                 # Cross-family transfer: a name diff can mean vendor-specific

modelexpress_client/python/modelexpress_rl/inference/adapter.py

@@ -9,6 +9,8 @@
 from dataclasses import dataclass
 from typing import Any
 
+from modelexpress import p2p_pb2
+from modelexpress.client import MxClientBase
 from modelexpress_rl.train import WeightPayloadFormat
 
 
@@ -59,6 +61,30 @@ def physical_fingerprint(self) -> tuple:
 class GeneratorEngineAdapter(ABC):
     """Engine-specific transfer planning and installation boundary."""
 
+    @property
+    @abstractmethod
+    def worker_rank(self) -> int:
+        """Return the rank used to match an inference P2P source."""
+
+    @abstractmethod
+    def build_p2p_identity(self, version_id: str) -> p2p_pb2.SourceIdentity:
+        """Build the engine-compatible P2P identity for an exact version."""
+
+    @abstractmethod
+    def stage_peer_weight(self, source: p2p_pb2.WorkerMetadata) -> object:
+        """Stage an exact version from a compatible inference peer."""
+
+    @abstractmethod
+    def publish_weight_version(
+        self,
+        *,
+        version_id: str,
+        staged: object,
+        p2p_client: MxClientBase,
+        worker_id: str,
+    ) -> None:
+        """Publish applied staging buffers as an exact-version P2P source."""
+
     @property
     @abstractmethod
     def supported_payload_formats(self) -> frozenset[WeightPayloadFormat]: