Refactor scheduler task concurrency evaluation into a dedicated helper.

[SameerMesiah97]

Description

This change refactors task concurrency evaluation logic out of SchedulerJobRunner._executable_task_instances_to_queued into a dedicated helper method.

Task concurrency checks for max_active_tis_per_dag and max_active_tis_per_dagrun, along with their associated serialized DAG loading, starvation bookkeeping, and missing DAG handling, were previously implemented inline inside the main scheduling loop. This logic has now been extracted into _task_concurrency_allows_execution.

The new helper encapsulates serialized DAG retrieval and missing DAG handling, task-level and task-per-DAG-run concurrency checks, and concurrency starvation bookkeeping and logging.

Rationale

_executable_task_instances_to_queued currently performs a large number of responsibilities, including pool management, DAG-level concurrency enforcement, task-level concurrency evaluation, executor routing, starvation tracking, and queue state transitions.

The task concurrency section in particular introduced deeply nested control flow and intermixed concurrency-specific behavior directly inside the main scheduling loop, making the scheduling path harder to visually scan and maintain.

Tests

No test adjustments were required because this change is a refactor only and does not introduce any behavioral changes.

Backwards Compatibility

This change is a refactor only and does not modify scheduler behavior, public APIs, task concurrency semantics, or executor interactions.

Was generative AI tooling used to co-author this PR?

• Yes (please specify the tool below)

Generated-by: [GPT 5.5] following the guidelines

[SameerMesiah97]

Requesting review for this.

[jroachgolf84]

@SameerMesiah97 - I think we should add some tests to these changes.

[SameerMesiah97]

@SameerMesiah97 - I think we should add some tests to these changes.

Could you please elaborate on this? Do you want me to add tests for the helper itself? I could of course do that but my understanding is that the concurrency behaviour has already been covered by the existing tests. Keep in mind that this PR is introducing no behavioural changes.

[SameerMesiah97]

Requesting review for this.

[SameerMesiah97]

Requesting review for this.

[steveahnahn]

IMO this doesn't need new tests, the path is already pinned.
Nit: the "avoid loading the serialized DAG unless required" comment now lives inside the helper, where the load is unconditional (the has_task_concurrency_limits guard stayed at the call site). Move it back to the caller.