[fully_async, one_step_off] feat: support auto resume on abort when using fully_async

[ArronHZG]

What does this PR do?

Refactor the fully_async code based on #5430 to support the gateway mode, and decouple the tool invocation and rollout processes during the partial rollout phase.

Feature use_trainer_do_validate is not ready for use, I will fix it in a subsequent PR.

Checklist Before Starting

• Search for similar PRs. Paste at least one query link here: ...
• Format the PR title as [{modules}] {type}: {description} (This will be checked by the CI)
  • {modules} include fsdp, megatron, veomni, sglang, vllm, rollout, trainer, ci, training_utils, recipe, hardware, deployment, ray, worker, single_controller, misc, perf, model, algo, env, tool, ckpt, doc, data, cfg, reward, fully_async, one_step_off
  • If this PR involves multiple modules, separate them with , like [megatron, fsdp, doc]
  • {type} is in feat, fix, refactor, chore, test
  • If this PR breaks any API (CLI arguments, config, function signature, etc.), add [BREAKING] to the beginning of the title.
  • Example: [BREAKING][fsdp, megatron] feat: dynamic batching

Test

For changes that can not be tested by CI (e.g., algorithm implementation, new model support), validate by experiment(s) and show results like training curve plots, evaluation results, etc.

API and Usage Example

Demonstrate how the API changes if any, and provide usage example(s) if possible.

# Add code snippet or script demonstrating how to use this

Design & Code Changes

Demonstrate the high-level design if this PR is complex, and list the specific changes.

Checklist Before Submitting

Important

Please check all the following items before requesting a review, otherwise the reviewer might deprioritize this PR for review.

• Read the Contribute Guide.
• Apply pre-commit checks: pre-commit install && pre-commit run --all-files --show-diff-on-failure --color=always
• Add / Update the documentation.
• Add unit or end-to-end test(s) to the CI workflow to cover all the code. If not feasible, explain why: ...
• Once your PR is ready for CI, send a message in the ci-request channel in the verl Slack workspace. (If not accessible, please try the Feishu group (飞书群).)
• If your PR is related to the recipe submodule, please also update the reference to the submodule commit via git submodule update --remote or cd recipe && git pull origin main.

[gemini-code-assist]

Code Review

This pull request introduces a significant refactoring of the fully_async policy to support automatic resumption of aborted rollouts. The changes simplify the architecture by removing the explicit cancel_queue and dedicated partial agent loops, integrating the retry logic directly into the FullyAsyncLLMServerManager. This refactoring also includes better code organization by moving utility functions and enhances robustness in asynchronous task handling through the new safe_create_task utility. The review comment highlights a critical aspect of robust asynchronous task handling by recommending proper exception logging in done_callbacks, which aligns with the goal of enhancing reliability.

[gemini-code-assist]

Raising an exception from a done_callback is not recommended. The exception is not propagated to the task's creator and is instead handled by the event loop's exception handler. This can lead to ungraceful process termination or hard-to-debug silent failures. It's better to log the exception with a full traceback here. If a failure in a task should stop the entire application, it should be done via a more explicit and graceful shutdown mechanism.

Suggested change

	except Exception as e:
	print(f"Task {task.get_name()} failed with exception: {e}")
	raise e
	except Exception:
	import logging
	import traceback
	logging.error(f"Task {task.get_name()} failed with exception:\n{traceback.format_exc()}")

[wuxibin89]

Do we still need a remote ParameterSynchronizer? Could it be absorb into FullyAsyncTrainer?

[ArronHZG]

Yes, it should work. The core logic has been moved over.

[Shangwei-Li]

Why should gpu_memory_utilization be set so low? It's preferred to set expandable memory instead of decrease gpu_memory_utilization:
export PYTORCH_NPU_ALLOC_CONF="expandable_segments:True"
+actor_rollout_ref.rollout.enable_sleep_mode=False

[ArronHZG]

just for ci oom,
Maybe I can submit a separate PR next time to modify the part you mentioned.

[PeterSH6]

Why changed the extra_info to extra_fields?

[ArronHZG]

Since AgentLoopOutput, AgentData, and _InternalAgentLoopOutput all use extra_fields as the variable for storing additional information, I have unified them here.

@PeterSH6