Fix scheduler refusing to start when [scheduler] only_idle is enabled (#76262)

The --only-idle argument read its default from config when the parser was built, so a config-enabled only_idle looked like an explicitly passed flag and tripped the guard requiring a positive --num-runs, which defaults to -1. It also made the option impossible to switch off from the command line.

Generated-by: Claude Opus 5

Co-authored-by: Eason09053360 <185830721+Eason09053360@users.noreply.github.com>

airflow-core/src/airflow/cli/cli_config.py

@@ -846,9 +846,13 @@ ARG_NUM_RUNS = Arg(
 
 ARG_ONLY_IDLE = Arg(
     ("-i", "--only-idle"),
-    default=conf.getboolean("scheduler", "only_idle", fallback=False),
-    help="Only count runs after the scheduler becomes idle.",
-    action="store_true",
+    default=None,
+    # Square brackets would be eaten as Rich markup by the help formatter, so name the option dotted.
+    help=(
+        "Only count runs after the scheduler becomes idle. Requires --num-runs to be a positive number. "
+        "Defaults to the scheduler.only_idle config option; pass --no-only-idle to override it."
+    ),
+    action=argparse.BooleanOptionalAction,
 )
 
 ARG_WITHOUT_MINGLE = Arg(

airflow-core/src/airflow/cli/commands/scheduler_command.py

@@ -37,13 +37,19 @@ from airflow.utils.scheduler_health import serve_health_check
 log = logging.getLogger(__name__)
 
 
+def _resolve_only_idle(args: Namespace) -> bool:
+    if args.only_idle is None:
+        return conf.getboolean("scheduler", "only_idle", fallback=False)
+    return args.only_idle
+
+
 @enable_memray_trace(component=MemrayTraceComponents.scheduler)
 def _run_scheduler_job(args) -> None:
     set_component_mp_start_method("scheduler")
     job_runner = SchedulerJobRunner(
         job=Job(),
         num_runs=args.num_runs,
-        only_idle=args.only_idle,
+        only_idle=_resolve_only_idle(args),
     )
     enable_health_check = conf.getboolean("scheduler", "ENABLE_HEALTH_CHECK")
     with _serve_logs(args.skip_serve_logs), _serve_health_check(enable_health_check):
@@ -56,8 +62,13 @@ def scheduler(args: Namespace):
     """Start Airflow Scheduler."""
     cli_utils.print_banner()
 
-    if args.only_idle and args.num_runs <= 0:
-        raise SystemExit("The --only-idle flag requires --num-runs to be set to a positive number.")
+    if _resolve_only_idle(args) and args.num_runs <= 0:
+        if args.only_idle:
+            raise SystemExit("The --only-idle flag requires --num-runs to be set to a positive number.")
+        log.warning(
+            "`[scheduler] only_idle` is enabled but has no effect because the run limit is not a "
+            "positive number; set --num-runs or `[scheduler] num_runs` to a positive number."
+        )
 
     if cli_utils.should_enable_hot_reload(args):
         from airflow.cli.hot_reload import run_with_reloader

airflow-core/src/airflow/config_templates/config.yml

@@ -2758,6 +2758,7 @@ scheduler:
       description: |
         Only count scheduler runs where the scheduler was idle (no tasks queued or finished)
         toward the run limit set by ``[scheduler] num_runs``. The count resets whenever a task is processed.
+        Has no effect unless ``[scheduler] num_runs`` is a positive number.
       version_added: 3.2.0
       type: boolean
       example: ~

airflow-core/tests/unit/cli/commands/test_scheduler_command.py

@@ -194,3 +194,41 @@ class TestSchedulerCommand:
         call_kwargs = mock_scheduler_job.call_args[1]
         assert call_kwargs["only_idle"] is True
         assert call_kwargs["num_runs"] == 5
+
+    @conf_vars({("scheduler", "only_idle"): "True"})
+    @mock.patch("airflow.cli.commands.scheduler_command.SchedulerJobRunner")
+    @mock.patch("airflow.cli.commands.scheduler_command.Process")
+    def test_only_idle_from_config_does_not_block_startup(self, mock_process, mock_scheduler_job, caplog):
+        mock_scheduler_job.return_value.job_type = "SchedulerJob"
+        args = self.parser.parse_args(["scheduler"])
+        assert args.num_runs <= 0
+
+        scheduler_command.scheduler(args)
+
+        assert mock_scheduler_job.call_args[1]["only_idle"] is True
+        assert (
+            "`[scheduler] only_idle` is enabled but has no effect because the run limit is not a "
+            "positive number; set --num-runs or `[scheduler] num_runs` to a positive number." in caplog
+        )
+
+    @conf_vars({("scheduler", "only_idle"): "True"})
+    @mock.patch("airflow.cli.commands.scheduler_command.SchedulerJobRunner")
+    @mock.patch("airflow.cli.commands.scheduler_command.Process")
+    def test_no_only_idle_overrides_config(self, mock_process, mock_scheduler_job):
+        mock_scheduler_job.return_value.job_type = "SchedulerJob"
+        args = self.parser.parse_args(["scheduler", "--no-only-idle"])
+
+        scheduler_command.scheduler(args)
+
+        assert mock_scheduler_job.call_args[1]["only_idle"] is False
+
+    @conf_vars({("scheduler", "only_idle"): "True"})
+    @mock.patch("airflow.cli.commands.scheduler_command.SchedulerJobRunner")
+    @mock.patch("airflow.cli.commands.scheduler_command.Process")
+    def test_only_idle_from_config_reaches_job_runner(self, mock_process, mock_scheduler_job):
+        mock_scheduler_job.return_value.job_type = "SchedulerJob"
+        args = self.parser.parse_args(["scheduler", "-n", "5"])
+
+        scheduler_command.scheduler(args)
+
+        assert mock_scheduler_job.call_args[1]["only_idle"] is True

airflow-core/tests/unit/cli/test_cli_parser.py

@@ -444,6 +444,27 @@ class TestCli:
         args = parser.parse_args([])
         assert args.test == 0
 
+    @pytest.mark.parametrize(
+        ("argv", "expected"),
+        [
+            # None, not a bool: the parser must not bake ``[scheduler] only_idle`` in at import time.
+            ([], None),
+            (["--only-idle"], True),
+            (["--no-only-idle"], False),
+        ],
+    )
+    def test_only_idle_flag_is_tri_state(self, argv, expected):
+        args = cli_parser.get_parser().parse_args(["scheduler", *argv])
+        assert args.only_idle is expected
+
+    def test_only_idle_help_survives_rich_markup_rendering(self):
+        """The help formatter renders Rich markup, which silently swallows ``[section]`` text."""
+        stdout = StringIO()
+        with contextlib.redirect_stdout(stdout), pytest.raises(SystemExit):
+            cli_parser.get_parser().parse_args(["scheduler", "--help"])
+
+        assert "scheduler.only_idle" in stdout.getvalue()
+
     def test_commands_and_command_group_sections(self):
         parser = cli_parser.get_parser()