[core] feat(rdt): enable driver-side ray.put with NIXL tensor transport

[n3sfan]

Allow ray.put(value, _tensor_transport='nixl') from a driver process. Previously this failed because get_runtime_context().current_actor raises RuntimeError in a non-actor context.

Changes:

• Make RDTMeta.src_actor and TransferMetadata.src_actor Optional
• Catch RuntimeError in put_object and set src_actor=None for driver
• Reject two-sided transports (NCCL/GLOO) from driver with clear error
• Guard src_actor dereferences in _abort_transport and transfer paths
• Handle free_object_primary_copy locally for driver-owned objects
• Raise ValueError for _use_object_store=True on driver-owned RDT objects
• Align worker.py error to RuntimeError with 'one-sided transport' message

Fixes #64714

Tests

Three new tests in python/ray/tests/rdt/test_rdt_nixl.py:

test_driver_put_nixl

End-to-end driver-side NIXL flow:

1. ray.put(tensor, _tensor_transport="nixl") from the driver
2. Verify src_actor is None in RDT metadata
3. Verify tensor registered in NIXL transport
4. Actor consumes the driver-created ref via actor.sum.remote(ref)
5. Driver-local ray.get(ref) returns the original tensor
6. After del ref + gc.collect(), verify cleanup: tensor freed, RDT metadata removed, NIXL metadata removed, tensor descriptor cache cleared

test_driver_owned_rdt_rejects_object_store_fallback

Verify _use_object_store=True on a driver-owned RDT object raises ValueError matching _use_object_store=True instead of crashing with AttributeError on None.__ray_call__.

test_driver_rejects_two_sided_transport

Verify ray.put(tensor, _tensor_transport="nccl") from the driver raises RuntimeError matching one-sided transport. Two-sided transports require an actor context to coordinate the send.

Running

# New driver tests
pytest -sv python/ray/tests/rdt/test_rdt_nixl.py \
  -k "test_driver_put_nixl or test_driver_rejects_two_sided_transport or test_driver_owned_rdt_rejects_object_store_fallback"

Results

Test	Result
test_driver_put_nixl	PASSED
test_driver_owned_rdt_rejects_object_store_fallback	PASSED
test_driver_rejects_two_sided_transport	PASSED

[gemini-code-assist]

Code Review

This pull request introduces support for driver-owned RDT (Ray Distributed Tensor) objects, allowing ray.put() to be called from the driver using one-sided transports like NIXL. It updates the RDT manager to handle cases where src_actor is None during transport abortion, actor cleanup, and object freeing, and adds corresponding unit tests. The review feedback suggests correcting a misleading error message in worker.py regarding two-sided transports and changing a ValueError to RuntimeError in rdt_manager.py for consistency.

[cursor]

Cursor Bugbot has reviewed your changes using default effort and found 1 potential issue.

^{Reviewed by Cursor Bugbot for commit 9f2d119. Configure here.}