Preserve legacy API options in multiprocessing workers

src/flake8/checker.py

@@ -61,16 +61,21 @@ def _mp_prefork(
         _mp = None
 
 
-def _mp_init(argv: Sequence[str]) -> None:
+def _mp_init(
+    argv: Sequence[str],
+    options_override: argparse.Namespace,
+) -> None:
     global _mp
 
     # Ensure correct signaling of ^C using multiprocessing.Pool.
     signal.signal(signal.SIGINT, signal.SIG_IGN)
 
     # for `fork` this'll already be set
     if _mp is None:
-        plugins, options = parse_args(argv)
-        _mp = plugins.checkers, options
+        plugins, _ = parse_args(argv)
+        # The legacy API can mutate options after parsing, so reparsing argv
+        # alone is not enough to reconstruct worker state on non-fork starts.
+        _mp = plugins.checkers, options_override
 
 
 def _mp_run(filename: str) -> tuple[str, Results, dict[str, int]]:
@@ -192,7 +197,11 @@ def report(self) -> tuple[int, int]:
     def run_parallel(self) -> None:
         """Run the checkers in parallel."""
         with _mp_prefork(self.plugins, self.options):
-            pool = _try_initialize_processpool(self.jobs, self.argv)
+            pool = _try_initialize_processpool(
+                self.jobs,
+                self.argv,
+                self.options,
+            )
 
         if pool is None:
             self.run_serial()
@@ -547,10 +556,15 @@ def check_physical_eol(
 def _try_initialize_processpool(
     job_count: int,
     argv: Sequence[str],
+    options: argparse.Namespace,
 ) -> multiprocessing.pool.Pool | None:
     """Return a new process pool instance if we are able to create one."""
     try:
-        return multiprocessing.Pool(job_count, _mp_init, initargs=(argv,))
+        return multiprocessing.Pool(
+            job_count,
+            _mp_init,
+            initargs=(argv, options),
+        )
     except OSError as err:
         if err.errno not in SERIAL_RETRY_ERRNOS:
             raise

tests/integration/test_checker.py

@@ -289,9 +289,14 @@ def test_acquire_when_multiprocessing_pool_can_initialize():
     This simulates the behaviour on most common platforms.
     """
     with mock.patch("multiprocessing.Pool") as pool:
-        result = checker._try_initialize_processpool(2, [])
+        options = mock.Mock()
+        result = checker._try_initialize_processpool(2, [], options)
 
-    pool.assert_called_once_with(2, checker._mp_init, initargs=([],))
+    pool.assert_called_once_with(
+        2,
+        checker._mp_init,
+        initargs=([], options),
+    )
     assert result is pool.return_value
 
 
@@ -308,9 +313,14 @@ def test_acquire_when_multiprocessing_pool_can_not_initialize():
     https://github.com/python/cpython/blob/4e02981de0952f54bf87967f8e10d169d6946b40/Lib/multiprocessing/synchronize.py#L30-L33
     """
     with mock.patch("multiprocessing.Pool", side_effect=ImportError) as pool:
-        result = checker._try_initialize_processpool(2, [])
+        options = mock.Mock()
+        result = checker._try_initialize_processpool(2, [], options)
 
-    pool.assert_called_once_with(2, checker._mp_init, initargs=([],))
+    pool.assert_called_once_with(
+        2,
+        checker._mp_init,
+        initargs=([], options),
+    )
     assert result is None
 
 

tests/unit/test_checker_manager.py

@@ -73,6 +73,23 @@ def test_jobs_count_limited_to_file_count():
     assert manager.jobs == 2
 
 
+def test_mp_init_preserves_supplied_options():
+    parsed_plugins = mock.Mock(checkers="parsed-checkers")
+    parsed_options = mock.Mock()
+    options_override = mock.Mock()
+
+    with (
+        mock.patch.object(checker, "_mp", None),
+        mock.patch.object(
+            checker,
+            "parse_args",
+            return_value=(parsed_plugins, parsed_options),
+        ),
+    ):
+        checker._mp_init([], options_override)
+        assert checker._mp == ("parsed-checkers", options_override)
+
+
 def test_make_checkers():
     """Verify that we create a list of FileChecker instances."""
     style_guide = style_guide_mock()