[fix](ray) Publish completion after stream retirement

[kaka11chen]

Summary

• keep a terminal stream record pending while RayStreamAdapter.retire() releases task/source ownership and deletes the ObjectRef stream
• publish the collector complete event only after that local retirement finishes
• replace the GC-timing assertion with a deterministic retirement gate that proves completion cannot overtake cleanup

Root cause

AsyncResultCollector._maybe_complete_record() removed the record and published complete before calling record.adapter.retire(). Consumers could therefore observe an empty collector while the adapter still held strong references to TaskLeaseObjectRefGenerator, making the weakref assertion scheduler-dependent in the full fast suite.

Testing

• python -m pytest tests/fast/test_udf_stream_result_collector.py (21 passed)
• related stream adapter and collector tests (33 passed)
• python -m pytest tests/fast/test_udf_process.py (6 passed)
• scripts/run_release_tests.sh (418 passed, 1 skipped; 7 real-Ray tests passed)
• scripts/run_fast_tests.sh non-ray (5895 passed, 28 skipped, 8 xfailed, 1 xpassed)
• pre-commit run --from-ref upstream/main --to-ref HEAD

Closes #217

[Copilot]

Copilot was unable to review this pull request because the user who requested the review has reached their quota limit.

[kaka11chen]

@codex review

[chatgpt-codex-connector]

Codex Review: Didn't find any major issues. 🚀

Reviewed commit: 190ef100e3

ℹ️ About Codex in GitHub

Your team has set up Codex to review pull requests in this repo. Reviews are triggered when you

• Open a pull request for review
• Mark a draft as ready
• Comment "@codex review".

If Codex has suggestions, it will comment; otherwise it will react with 👍.

Codex can also answer questions or update the PR. Try commenting "@codex address that feedback".

[QuakeWang]

+1