From 20d8b29c9cd621687825ae6a6c18e9000123fdf3 Mon Sep 17 00:00:00 2001
From: joncrall <erotemic@gmail.com>
Date: Wed, 21 Jan 2026 20:21:09 -0500
Subject: [PATCH 1/9] Prevent duplicate output in 3.14

---
 CHANGELOG.rst                      |  1 +
 line_profiler/explicit_profiler.py | 30 +++++++++++++++++++++++++++++-
 2 files changed, 30 insertions(+), 1 deletion(-)

diff --git a/CHANGELOG.rst b/CHANGELOG.rst
index 6daf8562..5969c53a 100644
--- a/CHANGELOG.rst
+++ b/CHANGELOG.rst
@@ -3,6 +3,7 @@ Changes
 
 5.0.1
 ~~~~~
+* FIX: Prevented duplicate or inconsistent profiler output under Python 3.14 when multiprocessing is used.
 * ENH: Add %%lprun_all for more beginner-friendly profiling in IPython/Jupyter #383
 * FIX: mitigate speed regressions introduced in 5.0.0
 * ENH: Added capability to combine profiling data both programmatically (``LineStats.__add__()``) and via the CLI (``python -m line_profiler``) (#380, originally proposed in #219)
diff --git a/line_profiler/explicit_profiler.py b/line_profiler/explicit_profiler.py
index bb7a904a..f505b14b 100644
--- a/line_profiler/explicit_profiler.py
+++ b/line_profiler/explicit_profiler.py
@@ -171,6 +171,12 @@ def func4():
 from .line_profiler import LineProfiler
 from .toml_config import ConfigSource
 
+# The first process that enables profiling records its PID here. Child processes
+# created via multiprocessing (spawn/forkserver) inherit this environment value,
+# allowing them to avoid registering duplicate atexit hooks (which can print
+# output after the parent exits and/or clobber output files).
+_OWNER_PID_ENVVAR = 'LINE_PROFILER_OWNER_PID'
+
 
 class GlobalProfiler:
     """
@@ -265,6 +271,7 @@ def __init__(self, config=None):
 
         self._profile = None
         self.enabled = None
+        self._owner_pid = None
 
         # Configs:
         # - How to toggle the profiler
@@ -310,6 +317,27 @@ def enable(self, output_prefix=None):
         """
         Explicitly enables global profiler and controls its settings.
         """
+        # When using multiprocessing start methods like 'spawn'/'forkserver',
+        # helper processes may import this module. We only register the atexit
+        # reporting hook (and enable profiling) in the first process that
+        # called enable(), to prevent duplicate/out-of-order output.
+        owner = os.environ.get(_OWNER_PID_ENVVAR)
+        if owner is None:
+            owner_pid = os.getpid()
+            os.environ[_OWNER_PID_ENVVAR] = str(owner_pid)
+        else:
+            try:
+                owner_pid = int(owner)
+            except Exception:
+                owner_pid = os.getpid()
+                os.environ[_OWNER_PID_ENVVAR] = str(owner_pid)
+        self._owner_pid = owner_pid
+
+        # Only enable + register atexit in the owner process.
+        if os.getpid() != owner_pid:
+            self.enabled = False
+            return
+
         if self._profile is None:
             # Try to only ever create one real LineProfiler object
             atexit.register(self.show)
@@ -366,8 +394,8 @@ def show(self):
         write_timestamped_text = self.write_config['timestamped_text']
         write_lprof = self.write_config['lprof']
 
+        kwargs = {'config': self._config, **self.show_config}
         if write_stdout:
-            kwargs = {'config': self._config, **self.show_config}
             self._profile.print_stats(**kwargs)
 
         if write_text or write_timestamped_text:

From 3d0f9817b26a08ab5e700492eb6296683dbfc0f8 Mon Sep 17 00:00:00 2001
From: Jon Crall <erotemic@gmail.com>
Date: Fri, 23 Jan 2026 20:33:43 -0500
Subject: [PATCH 2/9] Register explicit profiler atexit hook only in main
 process (fix multiprocessing / Py3.14 regression) (#5)

* Fix explicit profiler ownership for multiprocessing

* Avoid helper process atexit registration

* Skip atexit output in forked children

* Refine ownership checks for explicit profiler

* Add debug hooks and reduce CI matrix

* Skip orphaned forkserver output

* Restore full CI matrix
---
 line_profiler/explicit_profiler.py | 164 +++++++++++++++++++++++++----
 tests/test_complex_case.py         |   1 +
 tests/test_explicit_profile.py     |  82 +++++++++++++++
 3 files changed, 227 insertions(+), 20 deletions(-)

diff --git a/line_profiler/explicit_profiler.py b/line_profiler/explicit_profiler.py
index f505b14b..daff1ff9 100644
--- a/line_profiler/explicit_profiler.py
+++ b/line_profiler/explicit_profiler.py
@@ -164,7 +164,9 @@ def func4():
 The core functionality in this module was ported from :mod:`xdev`.
 """
 import atexit
+import multiprocessing
 import os
+import pathlib
 import sys
 # This is for compatibility
 from .cli_utils import boolean, get_python_executable as _python_command
@@ -173,8 +175,8 @@ def func4():
 
 # The first process that enables profiling records its PID here. Child processes
 # created via multiprocessing (spawn/forkserver) inherit this environment value,
-# allowing them to avoid registering duplicate atexit hooks (which can print
-# output after the parent exits and/or clobber output files).
+# which helps prevent helper processes from claiming ownership and clobbering
+# output. Standalone subprocess runs should always be able to reset this value.
 _OWNER_PID_ENVVAR = 'LINE_PROFILER_OWNER_PID'
 
 
@@ -270,9 +272,8 @@ def __init__(self, config=None):
         self._config = config_source.path
 
         self._profile = None
-        self.enabled = None
         self._owner_pid = None
-
+        self.enabled = None
         # Configs:
         # - How to toggle the profiler
         self.setup_config = config_source.conf_dict['setup']
@@ -317,27 +318,28 @@ def enable(self, output_prefix=None):
         """
         Explicitly enables global profiler and controls its settings.
         """
+        self._debug('enable:enter')
         # When using multiprocessing start methods like 'spawn'/'forkserver',
-        # helper processes may import this module. We only register the atexit
-        # reporting hook (and enable profiling) in the first process that
-        # called enable(), to prevent duplicate/out-of-order output.
-        owner = os.environ.get(_OWNER_PID_ENVVAR)
-        if owner is None:
-            owner_pid = os.getpid()
-            os.environ[_OWNER_PID_ENVVAR] = str(owner_pid)
-        else:
-            try:
-                owner_pid = int(owner)
-            except Exception:
-                owner_pid = os.getpid()
-                os.environ[_OWNER_PID_ENVVAR] = str(owner_pid)
-        self._owner_pid = owner_pid
+        # helper processes may import this module. Only register the atexit
+        # reporting hook (and enable profiling) in real script invocations to
+        # prevent duplicate/out-of-order output.
+        if self._is_helper_process_context():
+            self._debug('enable:helper-context')
+            self.enabled = False
+            return
 
-        # Only enable + register atexit in the owner process.
-        if os.getpid() != owner_pid:
+        if self._should_skip_due_to_owner():
+            self._debug('enable:skip-due-to-owner')
             self.enabled = False
             return
 
+        # Standalone script executions should always claim ownership, even if a
+        # PID marker was inherited from another process environment.
+        owner_pid = os.getpid()
+        os.environ[_OWNER_PID_ENVVAR] = str(owner_pid)
+        self._owner_pid = owner_pid
+        self._debug('enable:owner-claimed', owner_pid=owner_pid)
+
         if self._profile is None:
             # Try to only ever create one real LineProfiler object
             atexit.register(self.show)
@@ -350,6 +352,120 @@ def enable(self, output_prefix=None):
         if output_prefix is not None:
             self.output_prefix = output_prefix
 
+    def _is_helper_process_context(self):
+        """
+        Determine if this process looks like a multiprocessing helper.
+
+        Helper contexts should never register atexit hooks or claim ownership,
+        while real script invocations should always be allowed to do so.
+        """
+        argv0 = sys.argv[0] if sys.argv else ''
+        if self._has_forkserver_env():
+            self._debug('helper:forkserver-env', argv0=argv0)
+            return True
+        try:
+            import multiprocessing.spawn as mp_spawn
+            if getattr(mp_spawn, '_inheriting', False):
+                self._debug('helper:spawn-inheriting', argv0=argv0)
+                return True
+        except Exception:
+            pass
+        try:
+            if multiprocessing.current_process().name != 'MainProcess':
+                self._debug(
+                    'helper:non-main-process',
+                    process_name=multiprocessing.current_process().name,
+                    argv0=argv0,
+                )
+                return True
+        except Exception:
+            pass
+
+        main_mod = sys.modules.get('__main__')
+        main_file = getattr(main_mod, '__file__', None)
+        for candidate in (argv0, main_file):
+            if candidate:
+                try:
+                    if pathlib.Path(candidate).exists():
+                        self._debug('helper:script-detected', candidate=candidate)
+                        return False
+                except Exception:
+                    continue
+
+        self._debug('helper:no-script-detected', argv0=argv0, main_file=main_file)
+        return True
+
+    def _should_skip_due_to_owner(self):
+        """
+        In multiprocessing children, respect an inherited owner marker.
+
+        Standalone subprocesses (parent_process is None) should reset ownership,
+        but fork/spawn children should not clobber a parent owner's outputs.
+        """
+        try:
+            if multiprocessing.parent_process() is None:
+                self._debug('owner:no-parent', owner=os.environ.get(_OWNER_PID_ENVVAR))
+                return False
+        except Exception:
+            return False
+
+        owner = os.environ.get(_OWNER_PID_ENVVAR)
+        if owner is None:
+            return False
+
+        try:
+            owner_pid = int(owner)
+            if os.getppid() == 1 and owner_pid != os.getpid():
+                self._debug('owner:skip-orphan', owner=owner, ppid=os.getppid())
+                return True
+            if os.getppid() == owner_pid and owner_pid != os.getpid():
+                try:
+                    start_method = multiprocessing.get_start_method(allow_none=True)
+                except Exception:
+                    start_method = None
+                if start_method == 'forkserver':
+                    self._debug(
+                        'owner:skip-forkserver-child',
+                        owner=owner,
+                        ppid=os.getppid(),
+                        start_method=start_method,
+                    )
+                    return True
+            skip = owner_pid != os.getpid()
+            self._debug('owner:check', owner=owner, skip=skip)
+            return skip
+        except Exception:
+            return False
+
+    def _has_forkserver_env(self):
+        for key in os.environ:
+            if key.startswith('FORKSERVER_'):
+                return True
+            if key.startswith('MULTIPROCESSING_FORKSERVER'):
+                return True
+        return False
+
+    def _debug(self, message, **extra):
+        if not os.environ.get('LINE_PROFILER_DEBUG'):
+            return
+        try:
+            parent = multiprocessing.parent_process()
+            parent_pid = parent.pid if parent is not None else None
+        except Exception:
+            parent_pid = None
+        info = {
+            'pid': os.getpid(),
+            'ppid': os.getppid(),
+            'process': getattr(multiprocessing.current_process(), 'name', None),
+            'parent_pid': parent_pid,
+            'owner_env': os.environ.get(_OWNER_PID_ENVVAR),
+            'owner_pid': self._owner_pid,
+            'enabled': self.enabled,
+        }
+        info.update(extra)
+        payload = ' '.join(f'{k}={v!r}' for k, v in info.items())
+        print(f'[line_profiler debug] {message} {payload}')
+
     def disable(self):
         """
         Explicitly initialize and disable this global profiler.
@@ -386,6 +502,14 @@ def show(self):
         If the implicit setup triggered, then this will be called by
         :py:mod:`atexit`.
         """
+        self._debug('show:enter')
+        owner_env = os.environ.get(_OWNER_PID_ENVVAR)
+        if os.getppid() == 1 and owner_env == str(os.getpid()):
+            self._debug('show:skip-orphan-owner', owner_env=owner_env)
+            return
+        if self._owner_pid is not None and os.getpid() != self._owner_pid:
+            self._debug('show:skip-non-owner', current_pid=os.getpid())
+            return
         import io
         import pathlib
 
diff --git a/tests/test_complex_case.py b/tests/test_complex_case.py
index bd705929..5fa94580 100644
--- a/tests/test_complex_case.py
+++ b/tests/test_complex_case.py
@@ -95,6 +95,7 @@ def test_varied_complex_invocations():
             temp_dpath = stack.enter_context(tempfile.TemporaryDirectory())
             stack.enter_context(ub.ChDir(temp_dpath))
             env = {}
+            env['LINE_PROFILER_DEBUG'] = '1'
 
             outpath = case['outpath']
             if outpath:
diff --git a/tests/test_explicit_profile.py b/tests/test_explicit_profile.py
index 038e6582..f6b0da8c 100644
--- a/tests/test_explicit_profile.py
+++ b/tests/test_explicit_profile.py
@@ -141,6 +141,88 @@ def test_explicit_profile_with_environ_on():
         assert (temp_dpath / 'profile_output.lprof').exists()
 
 
+def test_explicit_profile_ignores_inherited_owner_marker():
+    """
+    Standalone runs should not be blocked by an inherited owner marker.
+    """
+    with tempfile.TemporaryDirectory() as tmp:
+        temp_dpath = ub.Path(tmp)
+        env = os.environ.copy()
+        env['LINE_PROFILE'] = '1'
+        env['LINE_PROFILER_OWNER_PID'] = str(os.getpid() + 100000)
+        env['PYTHONPATH'] = os.getcwd()
+
+        with ub.ChDir(temp_dpath):
+
+            script_fpath = ub.Path('script.py')
+            script_fpath.write_text(_demo_explicit_profile_script())
+
+            args = [sys.executable, os.fspath(script_fpath)]
+            proc = ub.cmd(args, env=env)
+            print(proc.stdout)
+            print(proc.stderr)
+            proc.check_returncode()
+
+        assert (temp_dpath / 'profile_output.txt').exists()
+        assert (temp_dpath / 'profile_output.lprof').exists()
+
+
+def test_explicit_profile_process_pool_forkserver():
+    """
+    Ensure explicit profiler works with forkserver ProcessPoolExecutor.
+    """
+    import multiprocessing as mp
+    if 'forkserver' not in mp.get_all_start_methods():
+        pytest.skip('forkserver start method not available')
+    with tempfile.TemporaryDirectory() as tmp:
+        temp_dpath = ub.Path(tmp)
+        env = os.environ.copy()
+        env['LINE_PROFILE'] = '1'
+        env['LINE_PROFILER_DEBUG'] = '1'
+        env['PYTHONPATH'] = os.getcwd()
+
+        with ub.ChDir(temp_dpath):
+
+            script_fpath = ub.Path('script.py')
+            script_fpath.write_text(ub.codeblock(
+                '''
+                import multiprocessing as mp
+                from concurrent.futures import ProcessPoolExecutor
+                from line_profiler import profile
+
+                def worker(x):
+                    return x * x
+
+                @profile
+                def run():
+                    total = 0
+                    for i in range(1000):
+                        total += i % 7
+                    with ProcessPoolExecutor(max_workers=2) as ex:
+                        list(ex.map(worker, range(4)))
+                    return total
+
+                def main():
+                    if 'forkserver' in mp.get_all_start_methods():
+                        mp.set_start_method('forkserver', force=True)
+                    run()
+
+                if __name__ == '__main__':
+                    main()
+                ''').strip())
+
+            args = [sys.executable, os.fspath(script_fpath)]
+            proc = ub.cmd(args, env=env)
+            print(proc.stdout)
+            print(proc.stderr)
+            proc.check_returncode()
+
+        output_path = temp_dpath / 'profile_output.txt'
+        assert output_path.exists()
+        assert output_path.stat().st_size > 100
+        assert proc.stdout.count('Wrote profile results to profile_output.txt') == 1
+
+
 def test_explicit_profile_with_environ_off():
     """
     When LINE_PROFILE is falsy, profiling should not run.

From e7151e55a06abc051d0e4dabf5bb0500aee86c61 Mon Sep 17 00:00:00 2001
From: joncrall <erotemic@gmail.com>
Date: Fri, 23 Jan 2026 22:01:50 -0500
Subject: [PATCH 3/9] Disable debug lines

---
 tests/test_complex_case.py     | 3 ++-
 tests/test_explicit_profile.py | 2 +-
 2 files changed, 3 insertions(+), 2 deletions(-)

diff --git a/tests/test_complex_case.py b/tests/test_complex_case.py
index 5fa94580..ace7502b 100644
--- a/tests/test_complex_case.py
+++ b/tests/test_complex_case.py
@@ -95,7 +95,8 @@ def test_varied_complex_invocations():
             temp_dpath = stack.enter_context(tempfile.TemporaryDirectory())
             stack.enter_context(ub.ChDir(temp_dpath))
             env = {}
-            env['LINE_PROFILER_DEBUG'] = '1'
+            # Can enable if this breaks again
+            # env['LINE_PROFILER_DEBUG'] = '1'
 
             outpath = case['outpath']
             if outpath:
diff --git a/tests/test_explicit_profile.py b/tests/test_explicit_profile.py
index f6b0da8c..511509b1 100644
--- a/tests/test_explicit_profile.py
+++ b/tests/test_explicit_profile.py
@@ -178,7 +178,7 @@ def test_explicit_profile_process_pool_forkserver():
         temp_dpath = ub.Path(tmp)
         env = os.environ.copy()
         env['LINE_PROFILE'] = '1'
-        env['LINE_PROFILER_DEBUG'] = '1'
+        # env['LINE_PROFILER_DEBUG'] = '1'
         env['PYTHONPATH'] = os.getcwd()
 
         with ub.ChDir(temp_dpath):

From 273cbcb76d9ff9e2bc8d8bcb556bf5af7c85eb60 Mon Sep 17 00:00:00 2001
From: joncrall <erotemic@gmail.com>
Date: Fri, 6 Feb 2026 19:42:58 -0500
Subject: [PATCH 4/9] simplify fix

---
 line_profiler/explicit_profiler.py | 194 ++++++++++++++---------------
 1 file changed, 92 insertions(+), 102 deletions(-)

diff --git a/line_profiler/explicit_profiler.py b/line_profiler/explicit_profiler.py
index daff1ff9..522d2c79 100644
--- a/line_profiler/explicit_profiler.py
+++ b/line_profiler/explicit_profiler.py
@@ -317,132 +317,61 @@ def _implicit_setup(self):
     def enable(self, output_prefix=None):
         """
         Explicitly enables global profiler and controls its settings.
+
+        Notes:
+            Multiprocessing start methods like 'spawn'/'forkserver' can create
+            helper/bootstrap interpreters that import this module. Those helpers
+            must not claim ownership or register an atexit hook, otherwise they can
+            clobber output from the real script process.
         """
-        self._debug('enable:enter')
-        # When using multiprocessing start methods like 'spawn'/'forkserver',
-        # helper processes may import this module. Only register the atexit
-        # reporting hook (and enable profiling) in real script invocations to
-        # prevent duplicate/out-of-order output.
-        if self._is_helper_process_context():
-            self._debug('enable:helper-context')
+        self._debug("enable:ENTER")
+
+        if is_mp_bootstrap():
+            self._debug("enable:skip-mp-bootstrap")
             self.enabled = False
             return
 
         if self._should_skip_due_to_owner():
-            self._debug('enable:skip-due-to-owner')
+            self._debug("enable:skip-due-to-owner")
             self.enabled = False
             return
 
-        # Standalone script executions should always claim ownership, even if a
-        # PID marker was inherited from another process environment.
         owner_pid = os.getpid()
         os.environ[_OWNER_PID_ENVVAR] = str(owner_pid)
         self._owner_pid = owner_pid
-        self._debug('enable:owner-claimed', owner_pid=owner_pid)
+        self._debug("enable:owner-claimed", owner_pid=owner_pid)
 
         if self._profile is None:
-            # Try to only ever create one real LineProfiler object
             atexit.register(self.show)
-            self._profile = LineProfiler()  # type: ignore
+            self._profile = LineProfiler()
 
-        # The user can call this function more than once to update the final
-        # reporting or to re-enable the profiler after it a disable.
         self.enabled = True
-
         if output_prefix is not None:
             self.output_prefix = output_prefix
 
-    def _is_helper_process_context(self):
+    def _should_skip_due_to_owner(self) -> bool:
         """
-        Determine if this process looks like a multiprocessing helper.
+        Return True if another process has already claimed ownership.
 
-        Helper contexts should never register atexit hooks or claim ownership,
-        while real script invocations should always be allowed to do so.
+        The first process to enable profiling records its PID in an env var.
+        Child interpreters can inherit that value; they must not steal ownership.
         """
-        argv0 = sys.argv[0] if sys.argv else ''
-        if self._has_forkserver_env():
-            self._debug('helper:forkserver-env', argv0=argv0)
-            return True
-        try:
-            import multiprocessing.spawn as mp_spawn
-            if getattr(mp_spawn, '_inheriting', False):
-                self._debug('helper:spawn-inheriting', argv0=argv0)
-                return True
-        except Exception:
-            pass
-        try:
-            if multiprocessing.current_process().name != 'MainProcess':
-                self._debug(
-                    'helper:non-main-process',
-                    process_name=multiprocessing.current_process().name,
-                    argv0=argv0,
-                )
-                return True
-        except Exception:
-            pass
-
-        main_mod = sys.modules.get('__main__')
-        main_file = getattr(main_mod, '__file__', None)
-        for candidate in (argv0, main_file):
-            if candidate:
-                try:
-                    if pathlib.Path(candidate).exists():
-                        self._debug('helper:script-detected', candidate=candidate)
-                        return False
-                except Exception:
-                    continue
-
-        self._debug('helper:no-script-detected', argv0=argv0, main_file=main_file)
-        return True
-
-    def _should_skip_due_to_owner(self):
-        """
-        In multiprocessing children, respect an inherited owner marker.
-
-        Standalone subprocesses (parent_process is None) should reset ownership,
-        but fork/spawn children should not clobber a parent owner's outputs.
-        """
-        try:
-            if multiprocessing.parent_process() is None:
-                self._debug('owner:no-parent', owner=os.environ.get(_OWNER_PID_ENVVAR))
-                return False
-        except Exception:
-            return False
-
         owner = os.environ.get(_OWNER_PID_ENVVAR)
-        if owner is None:
+        if not owner:
+            self._debug("owner:no-owner-env")
             return False
 
-        try:
-            owner_pid = int(owner)
-            if os.getppid() == 1 and owner_pid != os.getpid():
-                self._debug('owner:skip-orphan', owner=owner, ppid=os.getppid())
-                return True
-            if os.getppid() == owner_pid and owner_pid != os.getpid():
-                try:
-                    start_method = multiprocessing.get_start_method(allow_none=True)
-                except Exception:
-                    start_method = None
-                if start_method == 'forkserver':
-                    self._debug(
-                        'owner:skip-forkserver-child',
-                        owner=owner,
-                        ppid=os.getppid(),
-                        start_method=start_method,
-                    )
-                    return True
-            skip = owner_pid != os.getpid()
-            self._debug('owner:check', owner=owner, skip=skip)
-            return skip
-        except Exception:
+        current = str(os.getpid())
+        if owner == current:
+            self._debug("owner:is-us", owner=owner)
             return False
 
-    def _has_forkserver_env(self):
-        for key in os.environ:
-            if key.startswith('FORKSERVER_'):
-                return True
-            if key.startswith('MULTIPROCESSING_FORKSERVER'):
-                return True
+        if is_mp_bootstrap():
+            self._debug("owner:skip-mp-bootstrap", owner=owner, current=current)
+            return True
+
+        # Standalone run: allow this interpreter to become the owner.
+        self._debug("owner:allow-standalone-reset", owner=owner, current=current)
         return False
 
     def _debug(self, message, **extra):
@@ -453,8 +382,10 @@ def _debug(self, message, **extra):
             parent_pid = parent.pid if parent is not None else None
         except Exception:
             parent_pid = None
+
+        pid = os.getpid()
+
         info = {
-            'pid': os.getpid(),
             'ppid': os.getppid(),
             'process': getattr(multiprocessing.current_process(), 'name', None),
             'parent_pid': parent_pid,
@@ -464,7 +395,7 @@ def _debug(self, message, **extra):
         }
         info.update(extra)
         payload = ' '.join(f'{k}={v!r}' for k, v in info.items())
-        print(f'[line_profiler debug] {message} {payload}')
+        print(f'[line_profiler debug {pid=}] {message} {payload}')
 
     def disable(self):
         """
@@ -511,7 +442,6 @@ def show(self):
             self._debug('show:skip-non-owner', current_pid=os.getpid())
             return
         import io
-        import pathlib
 
         write_stdout = self.write_config['stdout']
         write_text = self.write_config['text']
@@ -553,6 +483,66 @@ def show(self):
                   + str(lprof_output_fpath))
 
 
+def is_mp_bootstrap() -> bool:
+    """
+    True when this interpreter invocation looks like multiprocessing
+    bootstrapping/plumbing, where we must not claim ownership / write outputs.
+
+    CommandLine:
+        xdoctest -m line_profiler.explicit_profiler is_mp_bootstrap
+
+    Example:
+        >>> import pytest
+        >>> if is_mp_bootstrap():
+        ...     pytest.skip('Cannot test mp bootstrap detection from within an mp bootstrap process')
+
+        >>> import subprocess, sys
+        >>> def _py_bool(code, *extra_argv):
+        ...     out = subprocess.check_output(
+        ...         [sys.executable, "-c", code, *extra_argv],
+        ...         text=True,
+        ...     )
+        ...     return out.strip() == "True"
+
+        >>> # Normal script-like invocation: should NOT look like MP bootstrapping.
+        >>> _py_bool("from line_profiler.explicit_profiler import is_mp_bootstrap; print(is_mp_bootstrap())")
+        False
+
+        >>> # Multiprocessing bootstraps often pass `--multiprocessing-*` args.
+        >>> # We can supply these as script args (after -c) so Python accepts them,
+        >>> # and `sys.orig_argv` will still include them.
+        >>> _py_bool("from line_profiler.explicit_profiler import is_mp_bootstrap; print(is_mp_bootstrap())",
+        ...          "--multiprocessing-fork")
+        True
+
+        >>> _py_bool("from line_profiler.explicit_profiler import is_mp_bootstrap; print(is_mp_bootstrap())",
+        ...          "--multiprocessing-spawn")
+        True
+    """
+    try:
+        import multiprocessing.spawn as mp_spawn
+        if getattr(mp_spawn, "_inheriting", False):
+            return True
+    except Exception:
+        pass
+
+    orig = getattr(sys, "orig_argv", None) or []
+    if any(a.startswith("--multiprocessing") for a in orig):
+        return True
+    if any("multiprocessing.forkserver" in a for a in orig):
+        return True
+    if any("multiprocessing.spawn" in a for a in orig):
+        return True
+
+    try:
+        if multiprocessing.current_process().name != "MainProcess":
+            return True
+    except Exception:
+        pass
+
+    return False
+
+
 # Construct the global profiler.
 # The first time it is called, it will be initialized. This is usually a
 # NoOpProfiler unless the user requested the real one.

From 6a9e86c5bb8a7db81230fc206a3b5f97926ac6df Mon Sep 17 00:00:00 2001
From: joncrall <erotemic@gmail.com>
Date: Fri, 6 Feb 2026 20:39:21 -0500
Subject: [PATCH 5/9] Fix doctest

---
 line_profiler/explicit_profiler.py | 44 +++++++++++++-----------------
 1 file changed, 19 insertions(+), 25 deletions(-)

diff --git a/line_profiler/explicit_profiler.py b/line_profiler/explicit_profiler.py
index 522d2c79..006f7528 100644
--- a/line_profiler/explicit_profiler.py
+++ b/line_profiler/explicit_profiler.py
@@ -488,35 +488,29 @@ def is_mp_bootstrap() -> bool:
     True when this interpreter invocation looks like multiprocessing
     bootstrapping/plumbing, where we must not claim ownership / write outputs.
 
-    CommandLine:
-        xdoctest -m line_profiler.explicit_profiler is_mp_bootstrap
-
     Example:
         >>> import pytest
         >>> if is_mp_bootstrap():
         ...     pytest.skip('Cannot test mp bootstrap detection from within an mp bootstrap process')
-
-        >>> import subprocess, sys
-        >>> def _py_bool(code, *extra_argv):
-        ...     out = subprocess.check_output(
-        ...         [sys.executable, "-c", code, *extra_argv],
-        ...         text=True,
-        ...     )
-        ...     return out.strip() == "True"
-
-        >>> # Normal script-like invocation: should NOT look like MP bootstrapping.
-        >>> _py_bool("from line_profiler.explicit_profiler import is_mp_bootstrap; print(is_mp_bootstrap())")
-        False
-
-        >>> # Multiprocessing bootstraps often pass `--multiprocessing-*` args.
-        >>> # We can supply these as script args (after -c) so Python accepts them,
-        >>> # and `sys.orig_argv` will still include them.
-        >>> _py_bool("from line_profiler.explicit_profiler import is_mp_bootstrap; print(is_mp_bootstrap())",
-        ...          "--multiprocessing-fork")
-        True
-
-        >>> _py_bool("from line_profiler.explicit_profiler import is_mp_bootstrap; print(is_mp_bootstrap())",
-        ...          "--multiprocessing-spawn")
+        >>> import sys, subprocess, textwrap
+        >>> code = textwrap.dedent(r'''
+        ... import multiprocessing as mp
+        ... from line_profiler.explicit_profiler import is_mp_bootstrap
+        ...
+        ... def child(q):
+        ...     q.put(is_mp_bootstrap())
+        ...
+        ... if __name__ == "__main__":
+        ...     ctx = mp.get_context("spawn")
+        ...     q = ctx.Queue()
+        ...     p = ctx.Process(target=child, args=(q,))
+        ...     p.start()
+        ...     val = q.get()
+        ...     p.join()
+        ...     print(val)
+        ... ''')
+        >>> out = subprocess.check_output([sys.executable, "-c", code], text=True).strip()
+        >>> out in {"True", "False"}
         True
     """
     try:

From 7f706c0c38dbf003575facaca36c2db64874a8f0 Mon Sep 17 00:00:00 2001
From: joncrall <erotemic@gmail.com>
Date: Fri, 6 Feb 2026 20:54:46 -0500
Subject: [PATCH 6/9] Add typing to explicit profiler while working on it

---
 line_profiler/explicit_profiler.py  | 59 +++++++++++++++++++++--------
 line_profiler/explicit_profiler.pyi | 33 ----------------
 2 files changed, 43 insertions(+), 49 deletions(-)
 delete mode 100644 line_profiler/explicit_profiler.pyi

diff --git a/line_profiler/explicit_profiler.py b/line_profiler/explicit_profiler.py
index 006f7528..324ce33b 100644
--- a/line_profiler/explicit_profiler.py
+++ b/line_profiler/explicit_profiler.py
@@ -163,21 +163,32 @@ def func4():
 
 The core functionality in this module was ported from :mod:`xdev`.
 """
+from __future__ import annotations
 import atexit
 import multiprocessing
 import os
 import pathlib
 import sys
+import typing
+from typing import Any, Callable, TypeVar
+
+if typing.TYPE_CHECKING:
+    from typing import cast
+
+
 # This is for compatibility
 from .cli_utils import boolean, get_python_executable as _python_command
 from .line_profiler import LineProfiler
 from .toml_config import ConfigSource
 
+F = TypeVar('F', bound=Callable[..., Any])
+ConfigArg = str | pathlib.PurePath | bool | None
+
 # The first process that enables profiling records its PID here. Child processes
 # created via multiprocessing (spawn/forkserver) inherit this environment value,
 # which helps prevent helper processes from claiming ownership and clobbering
 # output. Standalone subprocess runs should always be able to reset this value.
-_OWNER_PID_ENVVAR = 'LINE_PROFILER_OWNER_PID'
+_OWNER_PID_ENVVAR: str = 'LINE_PROFILER_OWNER_PID'
 
 
 class GlobalProfiler:
@@ -187,7 +198,7 @@ class GlobalProfiler:
     The :py:obj:`line_profile.profile` decorator is an instance of this object.
 
     Arguments:
-        config (Union[str, PurePath, bool, None]):
+        config (str | PurePath | bool | None):
             Optional TOML config file from which to load the
             configurations (see Attributes);
             if not explicitly given (= :py:data:`True` or
@@ -195,8 +206,7 @@ class GlobalProfiler:
             :envvar:`!LINE_PROFILER_RC` environment variable or looked
             up among the current directory or its ancestors.  Should all
             that fail, the default config file at
-            ``importlib.resources.path('line_profiler.rc', \
-'line_profiler.toml')`` is used;
+            ``importlib.resources.path('line_profiler.rc', 'line_profiler.toml')`` is used;
             passing :py:data:`False` disables all lookup and falls back
             to the default configuration
 
@@ -266,7 +276,17 @@ class GlobalProfiler:
         >>> self.show()
     """
 
-    def __init__(self, config=None):
+    _config: pathlib.PurePath | None
+    _profile: LineProfiler | None
+    _owner_pid: int | None
+    enabled: bool | None
+
+    setup_config: dict[str, list[str]]
+    write_config: dict[str, Any]
+    show_config: dict[str, Any]
+    output_prefix: str
+
+    def __init__(self, config: ConfigArg = None) -> None:
         # Remember which config file we loaded settings from
         config_source = ConfigSource.from_config(config)
         self._config = config_source.path
@@ -288,7 +308,7 @@ def __init__(self, config=None):
         # supplied `config`)
         self.show_config.pop('column_widths')
 
-    def _kernprof_overwrite(self, profile):
+    def _kernprof_overwrite(self, profile: LineProfiler) -> None:
         """
         Kernprof will call this when it runs, so we can use its profile object
         instead of our own. Note: when kernprof overwrites us we wont register
@@ -298,7 +318,7 @@ def _kernprof_overwrite(self, profile):
         self._profile = profile
         self.enabled = True
 
-    def _implicit_setup(self):
+    def _implicit_setup(self) -> None:
         """
         Called once the first time the user decorates a function with
         ``line_profiler.profile`` and they have not explicitly setup the global
@@ -314,7 +334,7 @@ def _implicit_setup(self):
         else:
             self.disable()
 
-    def enable(self, output_prefix=None):
+    def enable(self, output_prefix: str | None = None) -> None:
         """
         Explicitly enables global profiler and controls its settings.
 
@@ -374,7 +394,7 @@ def _should_skip_due_to_owner(self) -> bool:
         self._debug("owner:allow-standalone-reset", owner=owner, current=current)
         return False
 
-    def _debug(self, message, **extra):
+    def _debug(self, message: str, **extra: Any) -> None:
         if not os.environ.get('LINE_PROFILER_DEBUG'):
             return
         try:
@@ -385,7 +405,7 @@ def _debug(self, message, **extra):
 
         pid = os.getpid()
 
-        info = {
+        info: dict[str, Any] = {
             'ppid': os.getppid(),
             'process': getattr(multiprocessing.current_process(), 'name', None),
             'parent_pid': parent_pid,
@@ -397,13 +417,13 @@ def _debug(self, message, **extra):
         payload = ' '.join(f'{k}={v!r}' for k, v in info.items())
         print(f'[line_profiler debug {pid=}] {message} {payload}')
 
-    def disable(self):
+    def disable(self) -> None:
         """
         Explicitly initialize and disable this global profiler.
         """
         self.enabled = False
 
-    def __call__(self, func):
+    def __call__(self, func: F) -> F:
         """
         If the global profiler is enabled, decorate a function to start the
         profiler on function entry and stop it on function exit. Otherwise
@@ -424,9 +444,14 @@ def __call__(self, func):
             self._implicit_setup()
         if not self.enabled:
             return func
-        return self._profile(func)
+        assert self._profile is not None
 
-    def show(self):
+        wrapped = self._profile(func)
+        if typing.TYPE_CHECKING:
+            wrapped = cast(F, wrapped)
+        return wrapped
+
+    def show(self) -> None:
         """
         Write the managed profiler stats to enabled outputs.
 
@@ -448,14 +473,16 @@ def show(self):
         write_timestamped_text = self.write_config['timestamped_text']
         write_lprof = self.write_config['lprof']
 
-        kwargs = {'config': self._config, **self.show_config}
+        assert self._profile is not None
+
+        kwargs: dict[str, Any] = {'config': self._config, **self.show_config}
         if write_stdout:
             self._profile.print_stats(**kwargs)
 
         if write_text or write_timestamped_text:
             stream = io.StringIO()
             # Text output always contains details, and cannot be rich.
-            text_kwargs = {**kwargs, 'rich': False, 'details': True}
+            text_kwargs: dict[str, Any] = {**kwargs, 'rich': False, 'details': True}
             self._profile.print_stats(stream=stream, **text_kwargs)
             raw_text = stream.getvalue()
 
diff --git a/line_profiler/explicit_profiler.pyi b/line_profiler/explicit_profiler.pyi
deleted file mode 100644
index f777689b..00000000
--- a/line_profiler/explicit_profiler.pyi
+++ /dev/null
@@ -1,33 +0,0 @@
-from pathlib import PurePath
-from typing import Dict
-from typing import List
-from typing import Callable
-from typing import Union
-from _typeshed import Incomplete
-
-
-class GlobalProfiler:
-    setup_config: Dict[str, List[str]]
-    output_prefix: str
-    write_config: Dict[str, bool]
-    show_config: Dict[str, bool]
-    enabled: bool | None
-
-    def __init__(self,
-                 config: Union[str, PurePath, bool, None] = None) -> None:
-        ...
-
-    def enable(self, output_prefix: Incomplete | None = ...) -> None:
-        ...
-
-    def disable(self) -> None:
-        ...
-
-    def __call__(self, func: Callable) -> Callable:
-        ...
-
-    def show(self) -> None:
-        ...
-
-
-profile: Incomplete

From db19b7544468158175fee108480ae7aeb4d4e54a Mon Sep 17 00:00:00 2001
From: joncrall <erotemic@gmail.com>
Date: Fri, 6 Feb 2026 21:00:12 -0500
Subject: [PATCH 7/9] Fix type error

---
 line_profiler/explicit_profiler.py | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/line_profiler/explicit_profiler.py b/line_profiler/explicit_profiler.py
index 324ce33b..aa9473c1 100644
--- a/line_profiler/explicit_profiler.py
+++ b/line_profiler/explicit_profiler.py
@@ -174,6 +174,7 @@ def func4():
 
 if typing.TYPE_CHECKING:
     from typing import cast
+    ConfigArg = str | pathlib.PurePath | bool | None
 
 
 # This is for compatibility
@@ -182,7 +183,6 @@ def func4():
 from .toml_config import ConfigSource
 
 F = TypeVar('F', bound=Callable[..., Any])
-ConfigArg = str | pathlib.PurePath | bool | None
 
 # The first process that enables profiling records its PID here. Child processes
 # created via multiprocessing (spawn/forkserver) inherit this environment value,

From d4c7adf95caee90274cf789a22f8bbb11a2249c0 Mon Sep 17 00:00:00 2001
From: joncrall <erotemic@gmail.com>
Date: Fri, 6 Feb 2026 21:15:18 -0500
Subject: [PATCH 8/9] Fix typing

---
 line_profiler/explicit_profiler.py | 9 ++-------
 line_profiler/line_profiler.pyi    | 2 +-
 2 files changed, 3 insertions(+), 8 deletions(-)

diff --git a/line_profiler/explicit_profiler.py b/line_profiler/explicit_profiler.py
index aa9473c1..1063c223 100644
--- a/line_profiler/explicit_profiler.py
+++ b/line_profiler/explicit_profiler.py
@@ -170,10 +170,9 @@ def func4():
 import pathlib
 import sys
 import typing
-from typing import Any, Callable, TypeVar
+from typing import Any, Callable
 
 if typing.TYPE_CHECKING:
-    from typing import cast
     ConfigArg = str | pathlib.PurePath | bool | None
 
 
@@ -182,8 +181,6 @@ def func4():
 from .line_profiler import LineProfiler
 from .toml_config import ConfigSource
 
-F = TypeVar('F', bound=Callable[..., Any])
-
 # The first process that enables profiling records its PID here. Child processes
 # created via multiprocessing (spawn/forkserver) inherit this environment value,
 # which helps prevent helper processes from claiming ownership and clobbering
@@ -423,7 +420,7 @@ def disable(self) -> None:
         """
         self.enabled = False
 
-    def __call__(self, func: F) -> F:
+    def __call__(self, func: Callable) -> Callable:
         """
         If the global profiler is enabled, decorate a function to start the
         profiler on function entry and stop it on function exit. Otherwise
@@ -447,8 +444,6 @@ def __call__(self, func: F) -> F:
         assert self._profile is not None
 
         wrapped = self._profile(func)
-        if typing.TYPE_CHECKING:
-            wrapped = cast(F, wrapped)
         return wrapped
 
     def show(self) -> None:
diff --git a/line_profiler/line_profiler.pyi b/line_profiler/line_profiler.pyi
index 7493dc58..0a9bdbb0 100644
--- a/line_profiler/line_profiler.pyi
+++ b/line_profiler/line_profiler.pyi
@@ -117,7 +117,7 @@ class LineProfiler(CLineProfiler, ByCountProfilerMixin):
     # Fallback: just wrap the `.__call__()` of a generic callable
 
     @overload
-    def __call__(self, func: Callable) -> FunctionType:
+    def __call__(self, func: Callable) -> Callable:
         ...
 
     def add_callable(

From 7e6e9d3ac9b18745ac66a23e97e2c8cb3578e294 Mon Sep 17 00:00:00 2001
From: joncrall <erotemic@gmail.com>
Date: Fri, 6 Feb 2026 21:51:26 -0500
Subject: [PATCH 9/9] wip

---
 line_profiler/explicit_profiler.py | 1 +
 1 file changed, 1 insertion(+)

diff --git a/line_profiler/explicit_profiler.py b/line_profiler/explicit_profiler.py
index 1063c223..787c4ac2 100644
--- a/line_profiler/explicit_profiler.py
+++ b/line_profiler/explicit_profiler.py
@@ -511,6 +511,7 @@ def is_mp_bootstrap() -> bool:
     bootstrapping/plumbing, where we must not claim ownership / write outputs.
 
     Example:
+        >>> # xdoctest: +SKIP('can be flaky at test time')
         >>> import pytest
         >>> if is_mp_bootstrap():
         ...     pytest.skip('Cannot test mp bootstrap detection from within an mp bootstrap process')