[core] feat: support async scheduling with structured outputs

[AlpinDale]

No description provided.

[gemini-code-assist]

Code Review

This pull request introduces a significant refactoring to support asynchronous scheduling with structured outputs. The core change involves splitting the model execution into a two-step process: execute_model for the forward pass and sample_tokens for sampling. This is a clean approach that allows for grammar bitmask computation between the two steps. The changes are consistently applied across the scheduler, engine core, executors, and model runners. I've found one critical issue that could lead to a crash in distributed setups.

[gemini-code-assist]

The assertion assert model_runner_output is not None can fail. The outputs list can contain None values from workers that did not produce an output, which is a valid scenario with the new changes. This assertion will cause a crash in distributed setups. You should handle None values gracefully by skipping them.

Suggested change

	assert model_runner_output is not None
	kv_output = model_runner_output.kv_connector_output
	if not kv_output:
	continue
	if model_runner_output is None:
	continue
	kv_output = model_runner_output.kv_connector_output
	if not kv_output:
	continue