refactor(rollout): dedup async driver plumbing; extract ToolAgentHarness

[haonan3]

Summary

Follow-up to #287, three self-contained commits, deliberately minimal against main (13 files, +396/−226; the agentic trainers are byte-identical to main):

Commit 1 — handle_fn shares its halves with the nowait seam. _launch_half (dispatch→localize→execute) and _collect_half (ray.get→rebind→collect) extracted on Handle; handle_fn, launch_nowait, and PendingHandleCall.result sequence the same two methods — the "keep the halves in parity" comment convention (#287's own listed follow-up) is deleted. Grad-context handling stays in handle_fn; no behavior change.

Commit 2 — single-copy launch clamp + enforced double-pull guard. Two minimal driver-side fixes, no loop moves:

• launch_ceiling(): the on-policy launch-clamp arithmetic both batch trainers carried as duplicated inline math becomes one module-level function. Ownership is stated in its docstring: trainer POLICY (rollout_id/sync_interval/num_rollouts vocabulary), never called by the engine classes, hosted in engine/asynchronous.py only as the trainers' one shared torch-free home. The batch trainers change by 2 lines each; their _next_step loops are otherwise untouched.
• AsyncAgenticRolloutEngine.submit() now enforces its documented precondition: a second submit while the prior drive is live raises instead of double-pulling the coordinator queue (_drive_live flag; reset when finalize_if_drained reports done and by quiesce). Agentic trainer loops untouched.

Commit 3 — ToolAgentHarness: task control flow as a worker-side plugin (harness split, step 1). The sequence a task runs by (turns, stop conditions, teardown) is dictated by task semantics — a third sequence owner next to training policy (trainer loops) and engine contracts — and now has its own home, inside the rollout worker, so multi-turn env state and intermediates never cross to the driver:

• unirl/rollout/harness/: protocol.py — RolloutHarness / HarnessContext (named engines + a suspend probe; no residency verbs — engine wake/sleep is the hosting runtime's deployment concern — and no trainer vocabulary) / HarnessOutcome (completed|suspended|failed + optional env_reward; the suspended doc is honest: resumability is the ENV's property — stateless envs resume, stateful envs pair with tail_policy: drop; config-time validation of that pairing is deferred to the config-selected-harness step). tool_agent.py — the generate → env.step → observe loop ported statement-for-statement from AgenticRolloutEngine._run_one (resume-aware turn budget, turn-boundary suspension, env-reward capture, failed-with-partial-trace fault isolation, guaranteed env.close teardown). Harness modules import ray/torch-free.
• AgenticRolloutEngine constructs the harness from its existing config fields (no recipe change); _run_one delegates with an EXHAUSTIVE status match inside the try — an unknown status or malformed outcome from a plugin is NaN-marked instead of slipping into training as completed or sinking the drain.
• Deletes the dead AgentLoop prototype (loop/agent_loop.py) and its RolloutEnginePort (loop/engine_port.py) — zero live callers; the loop now exists exactly once. loop/ keeps environments and tools.

Next steps (separate PRs): config-selected harness on the runtime; a second, genuinely different multimodal harness to validate the boundary; then migrate ComposedRolloutEngine's PE flow and delete it.

Related Issue

N/A (follow-up to #287; harness direction per the team architecture discussion).

Test Plan

Static (branch @ HEAD, base main @ 8d77a54): ruff format --check + ruff check over all changed files — clean; python3 -m py_compile — clean; lint/check_recipe_targets.py → 2434 recipe _target_ paths resolve (no recipe/config change anywhere); lint/check_experimental_boundaries.py → ok; stale-reference sweeps (AgentLoop|RolloutEnginePort|engine_port, and the reverted collect()/launch_until/poll_drive drafts) → zero hits.

CPU harnesses (uncommitted per tests-tree policy) — 20 passed:

• Driver plumbing (13): ray/torch-free import of engine.asynchronous; buffer freshest-first + staleness eviction + pop_evicted; pool complete-or-nothing retry + immediate KeyboardInterrupt; launch_ceiling values (rollout_id=3, interval=2, stale=1 → 6; stale=0, interval=1 → rollout_id+1; num_rollouts clamp); the trainer-shaped launch-then-reap loop over the verbs; launch- vs completion-time version stamping; submit double-pull guard lifecycle (second submit raises while live; finalize_if_drained done-report re-arms; quiesce re-arms); quiesce fold (abort + mandatory follow-up poll, pre-bump stamping); PendingGroups assembly/discard.
• ToolAgentHarness (7): ray/torch-free import of harness/*; runs until env done with last-reward-wins capture + teardown; max_turns + resume-awareness; suspension exactly at the turn boundary (in-flight turn finishes; teardown still runs); env fault → failed with partial trace preserved, partial reward dropped, teardown still runs; unknown engine name → KeyError listing provided names; frozen outcome.

The _run_one exhaustive-status net is not CPU-testable (engine imports torch); it is review-verified and exercised by the ALFWorld smoke below.

GPU end-to-end: Not run; reason: CPU-only dev box. Before undraft: async AR (examples/ar/qwen3_grpo_4b_base_dapo_sglang_async.yaml, expect ratio≈1) + BAGEL async diffusion (examples/diffusion/bagel/bagel_vllmomni_async.yaml, expect replay ratio=1.0000, ~8s/rollout) + one ALFWorld partial/async agentic smoke (exercises the submit guard and the harness delegation: suspension/carry, env reward, NaN-failure path).

Compatibility / Risk

• Nothing renamed, no recipe/config/knob changes; the only deleted symbols (AgentLoop, RolloutEnginePort) had zero live callers (re-exports removed from loop/__init__; out-of-tree importers of those two names must update — refactor(rollout): engine-level sync/async contracts for async rollout #287 no-alias precedent).
• The agentic trainers are byte-identical to main; the batch trainers differ by 2 lines each (the ceiling call). The known pump-loop duplication across the agentic pair deliberately REMAINS (glue may be copied; only the invariant got a single copy) — revisiting it is out of scope by decision, not oversight.
• Riskiest spot: the _run_one → harness port — statement-for-statement, with the failure-path contract preserved (partial trace kept, partial env reward dropped, close-on-all-paths) and pinned by the 7 CPU tests.
• Overlap: fix(rollout): guard agentic sync and deterministic diffusion eval #294 touches engine/asynchronous.py + trainers (mechanical rebase either direction; version semantics unchanged here); feat(rollout): dynamic async rollout scheduler #273/feat(rollout): per-lane dynamic dispatch for AR rollout (sync + async) #289 evolve the AR trainer loop, untouched here; refactor(trainer): extract the sync-loop template into BaseTrainer #282 disjoint.

Reviewer Notes

• Suggested order: commit-by-commit — handle.py halves (pure extraction) → asynchronous.py (+launch_ceiling, +guard) with the two 2-line trainer hunks → harness/protocol.py (the boundary and its rules) → harness/tool_agent.py vs the deleted _run_one body (side-by-side diff recommended) → engine.py delegation.
• Review-round record (second-model pass + lead): an earlier HEAD hoisted the agentic pump into an engine collect() with refill/on_evicted callbacks — rejected (trainer policy + callback inversion in the engine; the parameter-list vocabulary test is the guard) and reverted to main's trainer loops with only the guard kept engine-side; launch_until was dropped (2 lines of glue don't justify a cross-layer build callback); launch_ceiling stays in engine/asynchronous.py over a trainer-side home deliberately (a torch-heavy trainer module would kill the invariant's CPU numeric tests; a one-function module is ceremony) — its docstring carries the ownership statement; a PR split of the handle commit was made and then folded back per lead decision (commit-level separation suffices for review).
• AI-assisted (Claude Code); duplicate-work checks done (fix(rollout): guard agentic sync and deterministic diffusion eval #294/feat(rollout): per-lane dynamic dispatch for AR rollout (sync + async) #289/feat(rollout): dynamic async rollout scheduler #273/refactor(trainer): extract the sync-loop template into BaseTrainer #282).
• Draft until the GPU validation trio is run.

Checklist

• I reviewed the changed code and removed unrelated/generated artifacts.
• I updated tests, docs, and configs where needed, or explained why not.