[0z5a]

Stream accepted complete text-GRPO groups through real TransferQueue 0.1.9 CPU SimpleStorage before VIME's existing fixed-round training plan. Unique job/epoch/group partitions prevent initial puts from overwriting other groups. Generation, reward/filter and Sample conversion retain their existing ownership.

Review this feature: incremental diff (20 files, +1,855/-33). Prerequisite: maintenance #460 at a5b3e0b1. Target: 0z5a-branch at 3e569314 (merged #459); head b6f6c30a. The Files changed tab includes maintenance. History is preserved without force-push. #242/#265/#320 are references; this covers a limited part of #227.

The CI follow-up changes two production files (+24/-11). It restores the public four-parameter generate_rollout plugin contract; the queue callback uses a separate entry point only when TransferQueue is enabled.

The driver seals its batch, pauses admission/engines and confirms idle before training. READY→LEASED→PREPARED→TRAINING→TRAINED→COMMITTED is owned by VIME; native storage reads are not optimizer acknowledgments. A separate heartbeat covers the whole training plan. Owned partitions are deleted before capacity release; the whole serving cohort acknowledges a changed version before the next round. Checkpoints combine native model/optimizer/data state, acknowledged publication and a quiescent queue cursor.

Supported: standard synchronous text GRPO, TP=PP=CP=DP=1, no VPP, checkpointable global dataset and bounded deadlines/capacity. Default off imports no TQ and starts no extra services. Unsupported/debug/partial/critic/multimodal/MoE paths fail early. Sample telemetry, masks, logprobs, IDs and ragged top-p data round-trip unchanged. Restarts require the same job, common cursor/configuration and a larger epoch. Expired uncertain training becomes UNKNOWN and stops; no automatic optimizer replay or cross-crash exactly-once claim. Failed committed deletion retains GC without retraining. Working-set limits include transport copies and decoded samples, not the whole process RSS.

Actual validation	Result	Scope
Final stacked CPU suite	206 passed, 0 skipped	Codec, TTL/fencing, UNKNOWN, fixed-plan commit, configuration and shared regressions
Real SDK/controller/storage/client	1 passed, 0 skipped	101 rounds × 2 groups; concurrent writes; native put/get/delete; cursor restore; read/write/delete failure cases
Transport counters	PASS	205 successful puts, 204 gets, 337,400 payload bytes; peak working set 78,916 bytes ≤ 1 MiB
A100 Python 3.11 queue regression	35 passed	26 contract and 9 adapter tests; real TransferQueue 0.1.9 installed
Rollout signature follow-up	6 contract + 63 rollout tests passed	Previously reproducible 2 contract failures fixed; tested source hashes match the commit
Plugin CPU CI stage	All 27 files passed locally	Pipeline commands in an isolated Python 3.11 environment; two distributed files use a natural spawn/join wrapper
Qwen two-rank configuration check	PASS	32 float32 values; native asymmetric GPU visibility; both ranks NCCL 2.29.7 / NCCL_CUMEM_ENABLE=0, natural exit 0. Resource competition recorded; no optimizer update or speed claim
E0/E1: baseline/off/TQ model parity	BLOCKED	Actual model loss/advantage/gradient/parameter comparison pending
E2: ≥3 real generation→TQ→optimizer→publication rounds	BLOCKED	Backend fixture events are not model E2E
E3: fresh-process model/data/queue recovery + failures	BLOCKED	Backend cursor restore passed; real model/ack/publication failures pending
Maintenance CPU CI follow-up	PASS locally	Python 3.11 plugin/agent/utils stages and all 33 upstream-sync CPU files; fixed the stale Ouro loader list
Formatting / diff checks	PASS	Pinned all-file hooks and the two-file follow-up pass
Hosted CI #1407	PASS automated checks	GitHub success; Buildkite reports passed and blocked, so manual stages remain unexecuted. DCO and docs passed. Previous #1406 failed; authenticated step logs unavailable. The plugin signature failure was independently reproduced locally and fixed; no manual rerun

Thor backend checks use existing Python 3.12.3, Ray 2.58.0 and TransferQueue 0.1.9. The same real SDK/controller/storage/client check also passed on A100 (101 rounds × 2 groups), reusing the existing Ray service with voluntary actor exits. Its optimizer/publication events are fixtures. The A100 task environment includes router, TMS, TransferQueue/TensorDict and compiled Transformer Engine; shared Python/CUDA packages and drivers are unchanged. Qwen has not completed a native optimizer update: the initial NCCL failure showed different cuMem settings in the manual external engine and trainer. An aligned configuration is queued for validation under the shared GPU lock. Current Ouro collector/native results do not validate this queue path. The rollout-state field uses its concrete DataSource type.

Matched model experiment, three paired runs	Existing step	TQ step	Existing/TQ ratio	Result
Existing synchronous rollout vs strict-version TQ	—	—	—	NOT_RUN; native model gate pending

CPU put/get spans are transport measurements, not training throughput. No model speedup is claimed.

python -m pytest -q tests/rollout/test_transfer_queue_contract.py \
  tests/rollout/test_transfer_queue_adapter.py tests/test_megatron_argument_validation.py \
  tests/test_vllm_rollout.py tests/test_server_idle_contract.py tests/test_docs_consistency.py
python -m pytest -q -s tests/integration/test_transfer_queue_backend.py

The expiry-on-acquire, expiry-on-operation and regression review cases are covered. Remains Draft until model E0–E3, matched timings and CI pass. Raw logs, model files and coordination records stay outside the diff.

[read-the-docs-community]

Documentation build overview

📚 vime | 🛠️ Build #35298802 | 📁 Comparing 18bd03e against latest (791f101)

  🔍 Preview build  

48 files changed · + 3 added · ± 45 modified

+ Added

• advanced/draft-feature-collection.html
• advanced/transferqueue.html
• _examples_synced/ouro/README.html

± Modified

• genindex.html
• index.html
• search.html
• advanced/arch-support-beyond-megatron.html
• advanced/delta-weight-sync.html
• advanced/external-rollout-engines.html
• advanced/fault-tolerance.html
• advanced/low-precision.html
• advanced/megatron-config.html
• advanced/observability.html
• and 35 more...

[gemini-code-assist]

Code Review

This pull request introduces the TransferQueue streaming scope documentation, a local coordinator state contract (GroupLedger and associated dataclasses), and unit tests to validate lease acquisition, preparation, training, and expiration behaviors. The review feedback suggests automatically triggering lease expiration checks during acquire and _check operations to prevent stale states, along with adding a unit test to verify this auto-expiration logic.

[gemini-code-assist]

To prevent the ledger from getting stuck in a stale LEASED or PREPARED state when a lease expires, we should automatically trigger self.expire(now) at the beginning of acquire. This ensures that any expired lease is reclaimed immediately when a new worker attempts to acquire the group, without requiring the coordinator to explicitly call expire beforehand.

Suggested change

	def acquire(self, owner: str, token: str, now: float, ttl: float) -> Lease:
	if self.phase != Phase.READY or not owner or not token or ttl <= 0:
	raise ValueError("group is unavailable or lease arguments are invalid")
	self.generation += 1
	self.lease = Lease(owner, token, self.generation, now + ttl)
	self.phase = Phase.LEASED
	return self.lease
	def acquire(self, owner: str, token: str, now: float, ttl: float) -> Lease:
	self.expire(now)
	if self.phase != Phase.READY or not owner or not token or ttl <= 0:
	raise ValueError("group is unavailable or lease arguments are invalid")
	self.generation += 1
	self.lease = Lease(owner, token, self.generation, now + ttl)
	self.phase = Phase.LEASED
	return self.lease

[gemini-code-assist]

Similarly, calling self.expire(now) at the beginning of _check ensures that the ledger state is synchronized and transitioned to Phase.READY or Phase.UNKNOWN immediately when an expired lease is presented during operations like prepare, start_training, or commit.

Suggested change

	def _check(self, lease: Lease, now: float) -> None:
	if self.lease != lease or now >= lease.expires_at:
	raise ValueError("lease is stale or expired")
	def _check(self, lease: Lease, now: float) -> None:
	self.expire(now)
	if self.lease != lease or now >= lease.expires_at:
	raise ValueError("lease is stale or expired")

[gemini-code-assist]

Add a unit test to verify that the ledger automatically expires stale leases during acquire and other operations without requiring an explicit expire() call.

Suggested change

	def test_training_start_is_not_a_commit_boundary():
	ledger = GroupLedger(_group())
	lease = ledger.acquire("worker-a", "token-a", now=0, ttl=5)
	ledger.prepare(lease, "batch-1", now=1)
	ledger.start_training(lease, now=2)
	with pytest.raises(ValueError, match="not finished"):
	ledger.commit(lease, "early", now=3)
	ledger.finish_training(lease, now=4)
	ledger.expire(now=5)
	assert ledger.phase == Phase.UNKNOWN
	def test_training_start_is_not_a_commit_boundary():
	ledger = GroupLedger(_group())
	lease = ledger.acquire("worker-a", "token-a", now=0, ttl=5)
	ledger.prepare(lease, "batch-1", now=1)
	ledger.start_training(lease, now=2)
	with pytest.raises(ValueError, match="not finished"):
	ledger.commit(lease, "early", now=3)
	ledger.finish_training(lease, now=4)
	ledger.expire(now=5)
	assert ledger.phase == Phase.UNKNOWN
	def test_auto_expiration_on_acquire_and_check():
	ledger = GroupLedger(_group())
	old = ledger.acquire("worker-a", "token-a", now=0, ttl=5)
	ledger.prepare(old, "batch-1", now=4)
	new = ledger.acquire("worker-b", "token-b", now=5, ttl=10)
	assert ledger.phase == Phase.LEASED
	assert new.generation == old.generation + 1