[Bugfix] Don't reuse engine core payload buffer while zmq is sending it

[njhill]

EngineCoreProc.process_output_sockets recycles the msgpack payload bytearray across messages via MsgpackEncoder.encode_into, gated on the tracker returned by send_multipart(copy=False, track=True).

That gate never engages. Socket.send_multipart() returns a tracker for the last frame only, and pyzmq copies any frame smaller than zmq.COPY_THRESHOLD (64KiB), returning the always-done _FINISHED_TRACKER for it. The last frame is a small tensor buffer, so the tracker is always done and pending is dead code, while the payload frame is genuinely zero-copy once it exceeds 64KiB. The next encode_into then overwrites it mid-flight, and the client decodes a newer payload alongside the older message's tensor frames:

RuntimeError: shape '[29, 2]' is invalid for input of size 60
msgspec.ValidationError: cannot unpack non-iterable int object
    - at `$[1][69][3][0]`

Send the payload frame separately so the tracker covers the buffer we actually reuse. pyzmq's copy threshold then works in our favour: a small payload is copied and reused immediately, a large one is tracked properly. Also cap reuse_buffers when reclaiming from pending, which previously could not grow because that path was unreachable.

Seen intermittently in CI as v1/sample/test_logprobs_e2e.py failures, where lm_eval's large prompt-logprobs batches push the payload frame either side of the 64KiB threshold.

Also drop the retention of objects whose buffers were extracted for zero-copy send, both here and in MPClient.pending_messages. For a zero-copy frame pyzmq registers the buffer in its own gc registry (zmq/backend/cython/_zmq.py, _gc.store(data, ...)) and releases it only once libzmq is finished, and frames below zmq.COPY_THRESHOLD are copied outright. tensor_data() hands zmq a memoryview whose base chain reaches the source tensor, so refcounting - not timing - keeps the memory from being freed and reused. The retained references never covered device tensors anyway, since tensor_data() copies those to a temporary host tensor that the retained object does not reference. Neither client path reuses a send buffer, so that bookkeeping had no effect beyond looking like protection it did not provide.

[claude]

Claude Code Review

This pull request is from a fork — automated review is disabled. A repository maintainer can comment @claude review to run a one-time review.

[njhill]

CI failure unrelated, should be fixed by #50060

[AndreasKaratzas]

LGTM