fix: Fix kvbm nixl connector for vllm upgrade (#4957)

lib/bindings/kvbm/python/kvbm/vllm_integration/connector/pd_connector.py

@@ -12,7 +12,10 @@ from vllm.distributed.kv_transfer.kv_connector.v1.multi_connector import (
     MultiConnector,
     MultiKVConnectorMetadata,
 )
-from vllm.distributed.kv_transfer.kv_connector.v1.nixl_connector import NixlConnector
+from vllm.distributed.kv_transfer.kv_connector.v1.nixl_connector import (
+    NixlAgentMetadata,
+    NixlConnector,
+)
 from vllm.v1.core.sched.output import SchedulerOutput
 
 # Optional import for LMCache support
@@ -99,7 +102,31 @@ class PdConnector(MultiConnector):
         for c in self._connectors:
             c.set_xfer_handshake_metadata(metadata)
 
+    def get_handshake_metadata(self) -> KVConnectorHandshakeMetadata | None:
+        """
+        Get the KVConnector handshake metadata from the NIXL connector.
+
+        This metadata is used for out-of-band connector handshake between
+        P/D workers. The NIXL connector (second connector) provides this
+        metadata so decode workers can connect for KV transfer coordination.
+
+        Returns:
+            NixlAgentMetadata from the NIXL connector, or None if not available.
+        """
+        # Get handshake metadata from the NIXL connector (second connector)
+        nixl_connector = self._connectors[1]
+        metadata = nixl_connector.get_handshake_metadata()
+        if metadata is not None and not isinstance(metadata, NixlAgentMetadata):
+            raise TypeError(
+                f"Expected NixlAgentMetadata from NIXL connector, "
+                f"got {type(metadata).__name__}"
+            )
+        return metadata
+
     def bind_connector_metadata(self, connector_metadata: PdConnectorMetadata) -> None:
+        # Must call super() to set _connector_metadata so has_connector_metadata() returns True
+        # This is required for save_kv_layer to be called during the forward pass
+        super().bind_connector_metadata(connector_metadata)
         assert isinstance(connector_metadata, PdConnectorMetadata)
         if connector_metadata.extra_async_saves:
             self._extra_async_saves.update(connector_metadata.extra_async_saves)