[NSagan271]

What does this PR do?

Tensor uuids are globally unique, so the tensor store no longer keys by request. TensorStore keeps the tensors and which request owns each one (only for teardown). The reference state (refcounts, persist, mem_registered and the descriptor) moves behind a TensorBookkeeping interface that sees only integer uuids.

That interface has a Python backend and a Rust one (rust/src/tensors.rs). The Rust backend is a drop-in, covered by parity tests run against both, but nothing selects it by default: it copies descriptors in and rebuilds them on the way out, which pays off only for a Rust-side caller holding the same state (which we will have in the next few PRs of the stack).

The manager gains batched forms, each one boundary crossing under the Rust backend:

• store_and_return_tensor_info_batch returns flat columns for a whole output batch (this streamlines the back-and-forth with Rust, and will prevent building dataclasses when not needed)
• increment_ref_batch_uniform, dereference_batch(_uniform). The dereference function also handles the bookkeeping end of tensor cleanup (to reduce the number of round trips into Rust), and the downstream tensor cleanup is made aware of that.
• register_for_send_uuids, which the arena stages in one pass

The worker, data worker and graph nodes move onto these. The worker still routes per request in Python, so a small helper regroups the batch store's columns into each request's output edges.

cargo test now runs with --no-default-features (pyo3's extension-module feature is optional), and ParallelList joins utils/containers.

How was it tested?

(full testing to come after the entire stack is built)

Checklist

• ruff check . passes
• Added or updated tests / docs where relevant

[merceod]

Read the whole increment and ran it. cargo test 27/27 (the test binary needs the interpreter's lib dir on LD_LIBRARY_PATH locally, same as CI). Your parity tests plus 15 adversarial cases I added (mixed single and batched refcount paths, persist then drain then unpersist, one unknown uuid inside a batched call, columnar vs dict store on the same batch, batched cleanup leaving no ownership behind) all agree across the two backends. Suite 1267 passed with the one known failure, Bagel bitwise with main, storm clean. Good to go once #320 is in.

Nits. The worker relies on the store handing back the same descriptor object the edge carries so the arena's shm_segment stamp reaches the wire. That holds for the Python bookkeeper only, and not for a sharded clone even there. #325 fixes both with refresh_shm_placement, a one line pointer here would help whoever reads this PR on its own. increment_ref with a negative count is an assert in Python (gone under -O) and a ValueError in Rust, raising in both keeps the parity test honest. Outputs under a name no edge carries are no longer stored at all, where main stored and freed them at routing. Same memory outcome and it closes a leak for models that return extra tensors, worth a sentence in the description. And a line in rust/README about LD_LIBRARY_PATH for cargo test would save the next person the GLIBCXX hunt.

[Gaurav-Shah05]

cleanup=True pops the bookkeeping records before cleanup_collectable walks them, so one raising _cleanup_by_uuid (Mooncake's does) strands the rest of the batch with its records already gone. can_gc is then False so soft cleanup defers forever, and is_registered is False so force cleanup skips the unregistration.

[NSagan271]

Added a try / except so that raises in _cleanup_by_uuid don't strand the rest of the batch

[Gaurav-Shah05]

uuid_to_shard_dim is written per tensor but put_tensor_batch_* only runs after the loop, so a raise inside it leaves uuids there with nothing in the store, and neither cleanup path can reach them because both iterate get_all_uuids(rid).

[NSagan271]

Added a try / except around the loop, cleaning up uuid_to_shard_dim if an exception is raised

[Gaurav-Shah05]

nit: this implementation has no test, and truncating the loop to the first request passes the whole suite; of the three transports only the arena subclass is covered.

[Gaurav-Shah05]

nit: this re-names every accumulated segment on each growth rather than just the new one, so naming is quadratic in segment count, and the test's len(named) <= num_segments bound never exercises it because its 128-byte tensors don't grow the arena.

[NSagan271]

Now the implementation slices only the new tail of _seg_names, so it's no longer quadratic