From 3a4549b325c234fa8d533826a708fb0850dc4b0b Mon Sep 17 00:00:00 2001 From: HARSH THAKARE <151563948+harsh21234i@users.noreply.github.com> Date: Wed, 14 Oct 2026 14:59:47 +0530 Subject: [PATCH 01/11] fix: forward SIGINT to supervised command --- packages/reflex-base/src/reflex_base/utils/log.py | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/packages/reflex-base/src/reflex_base/utils/log.py b/packages/reflex-base/src/reflex_base/utils/log.py index 6a59344cbd7..93768a1b8aa 100644 --- a/packages/reflex-base/src/reflex_base/utils/log.py +++ b/packages/reflex-base/src/reflex_base/utils/log.py @@ -522,7 +522,9 @@ def supervise_output(args: list[str]) -> int: returncode = proc.wait() break except KeyboardInterrupt: - # The child gets the same interrupt and shuts down on its own. + # A PID-targeted SIGINT reaches only the supervisor. + if proc.poll() is None: + proc.send_signal(signal.SIGINT) continue wall_deadline = time.monotonic() + _DRAIN_WALL_SECONDS snapshots = [pump.writing_snapshot() for pump in pumps] From 1fee741047a56ad0e82e1785d300e8edd44bd846 Mon Sep 17 00:00:00 2001 From: HARSH THAKARE <151563948+harsh21234i@users.noreply.github.com> Date: Wed, 14 Oct 2026 15:00:17 +0530 Subject: [PATCH 02/11] test: cover PID-targeted SIGINT forwarding --- tests/units/reflex_base/utils/test_log.py | 16 ++++++++++++++++ 1 file changed, 16 insertions(+) diff --git a/tests/units/reflex_base/utils/test_log.py b/tests/units/reflex_base/utils/test_log.py index 0c062ffab36..43f27efa184 100644 --- a/tests/units/reflex_base/utils/test_log.py +++ b/tests/units/reflex_base/utils/test_log.py @@ -1167,6 +1167,22 @@ def test_supervise_output_forwards_sigterm(tmp_path): assert proc.wait(timeout=10) == 128 + signal.SIGTERM +@pytest.mark.skipif(sys.platform == "win32", reason="SIGINT is POSIX-only") +def test_supervise_output_forwards_pid_targeted_sigint(tmp_path): + """A PID-targeted interrupt reaches the supervised command.""" + script = tmp_path / "sigint.py" + script.write_text(_SIGTERM_SCRIPT) + with subprocess.Popen( + [sys.executable, str(script)], + stdout=subprocess.PIPE, + env={**os.environ, "REFLEX_LOG_JSON": "true"}, + ) as proc: + assert proc.stdout is not None + assert json.loads(proc.stdout.readline())["message"] == "ready" + proc.send_signal(signal.SIGINT) + assert proc.wait(timeout=10) == 128 + signal.SIGINT + + def test_output_pump_survives_a_gone_consumer(): """The reader keeps draining the child when its consumer closed the stream.""" child_read, child_write = os.pipe() From cef6350b570edf1311da31cb5b5e40031af519b9 Mon Sep 17 00:00:00 2001 From: HARSH THAKARE <151563948+harsh21234i@users.noreply.github.com> Date: Wed, 14 Oct 2026 15:00:46 +0530 Subject: [PATCH 03/11] news: document supervised SIGINT fix --- news/+7509-forward-sigint-to-supervised-process.bugfix.md | 1 + 1 file changed, 1 insertion(+) create mode 100644 news/+7509-forward-sigint-to-supervised-process.bugfix.md diff --git a/news/+7509-forward-sigint-to-supervised-process.bugfix.md b/news/+7509-forward-sigint-to-supervised-process.bugfix.md new file mode 100644 index 00000000000..ee7d0985478 --- /dev/null +++ b/news/+7509-forward-sigint-to-supervised-process.bugfix.md @@ -0,0 +1 @@ +`reflex run --json` now forwards SIGINT to its child process, so supervisors that signal only the Reflex PID can stop the app cleanly. From b1ab2a65d1911b5191bd939933047926673f504d Mon Sep 17 00:00:00 2001 From: HARSH THAKARE <151563948+harsh21234i@users.noreply.github.com> Date: Wed, 14 Oct 2026 15:24:19 +0530 Subject: [PATCH 04/11] fix: limit forwarded SIGINT to POSIX --- packages/reflex-base/src/reflex_base/utils/log.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/packages/reflex-base/src/reflex_base/utils/log.py b/packages/reflex-base/src/reflex_base/utils/log.py index 93768a1b8aa..79b46009866 100644 --- a/packages/reflex-base/src/reflex_base/utils/log.py +++ b/packages/reflex-base/src/reflex_base/utils/log.py @@ -522,8 +522,8 @@ def supervise_output(args: list[str]) -> int: returncode = proc.wait() break except KeyboardInterrupt: - # A PID-targeted SIGINT reaches only the supervisor. - if proc.poll() is None: + # A PID-targeted SIGINT reaches only the supervisor on POSIX. + if os.name == "posix" and proc.poll() is None: proc.send_signal(signal.SIGINT) continue wall_deadline = time.monotonic() + _DRAIN_WALL_SECONDS From ca153c80622366c2fdad9734e493fe9a59db51c5 Mon Sep 17 00:00:00 2001 From: Harsh Thakare Date: Wed, 14 Oct 2026 15:48:50 +0530 Subject: [PATCH 05/11] fix: avoid duplicate SIGINT during supervised shutdown --- ...ard-sigint-to-supervised-process.bugfix.md | 1 - packages/reflex-base/news/7558.bugfix.md | 1 + .../reflex-base/src/reflex_base/utils/log.py | 13 +++-- tests/units/reflex_base/utils/test_log.py | 48 +++++++++++++++++++ 4 files changed, 59 insertions(+), 4 deletions(-) delete mode 100644 news/+7509-forward-sigint-to-supervised-process.bugfix.md create mode 100644 packages/reflex-base/news/7558.bugfix.md diff --git a/news/+7509-forward-sigint-to-supervised-process.bugfix.md b/news/+7509-forward-sigint-to-supervised-process.bugfix.md deleted file mode 100644 index ee7d0985478..00000000000 --- a/news/+7509-forward-sigint-to-supervised-process.bugfix.md +++ /dev/null @@ -1 +0,0 @@ -`reflex run --json` now forwards SIGINT to its child process, so supervisors that signal only the Reflex PID can stop the app cleanly. diff --git a/packages/reflex-base/news/7558.bugfix.md b/packages/reflex-base/news/7558.bugfix.md new file mode 100644 index 00000000000..6e6053f0c27 --- /dev/null +++ b/packages/reflex-base/news/7558.bugfix.md @@ -0,0 +1 @@ +On POSIX, `reflex run --json` forwards SIGINT to its child process, so supervisors that signal only the Reflex PID can stop the app cleanly. diff --git a/packages/reflex-base/src/reflex_base/utils/log.py b/packages/reflex-base/src/reflex_base/utils/log.py index 79b46009866..b86fde6121a 100644 --- a/packages/reflex-base/src/reflex_base/utils/log.py +++ b/packages/reflex-base/src/reflex_base/utils/log.py @@ -507,7 +507,13 @@ def supervise_output(args: list[str]) -> int: # blocked on it waits for that read on Windows. out_read, out_write = os.pipe() err_read, err_write = os.pipe() - proc = subprocess.Popen(args, stdout=out_write, stderr=err_write, env=env) + proc = subprocess.Popen( + args, + stdout=out_write, + stderr=err_write, + env=env, + start_new_session=os.name == "posix", + ) os.close(out_write) os.close(err_write) pumps = [ @@ -522,9 +528,10 @@ def supervise_output(args: list[str]) -> int: returncode = proc.wait() break except KeyboardInterrupt: - # A PID-targeted SIGINT reaches only the supervisor on POSIX. + # Keep terminal SIGINT from reaching the child before we forward it. if os.name == "posix" and proc.poll() is None: - proc.send_signal(signal.SIGINT) + with contextlib.suppress(ProcessLookupError): + os.killpg(proc.pid, signal.SIGINT) continue wall_deadline = time.monotonic() + _DRAIN_WALL_SECONDS snapshots = [pump.writing_snapshot() for pump in pumps] diff --git a/tests/units/reflex_base/utils/test_log.py b/tests/units/reflex_base/utils/test_log.py index 43f27efa184..bc214f7211f 100644 --- a/tests/units/reflex_base/utils/test_log.py +++ b/tests/units/reflex_base/utils/test_log.py @@ -1183,6 +1183,54 @@ def test_supervise_output_forwards_pid_targeted_sigint(tmp_path): assert proc.wait(timeout=10) == 128 + signal.SIGINT +_SIGINT_CLEANUP_SCRIPT = """ +import signal +import sys +import time +from pathlib import Path + +from reflex_base.utils import log + +if not log.is_output_supervised(): + sys.exit(log.supervise_output([sys.executable, __file__, sys.argv[1]])) + +signal_count = 0 + +def handle_sigint(*_): + global signal_count + signal_count += 1 + +signal.signal(signal.SIGINT, handle_sigint) +print("ready", flush=True) +while signal_count == 0: + time.sleep(0.01) +time.sleep(0.25) +Path(sys.argv[1]).write_text(str(signal_count)) +""" + + +@pytest.mark.skipif(sys.platform == "win32", reason="SIGINT is POSIX-only") +def test_supervise_output_does_not_double_forward_terminal_sigint(tmp_path): + """A terminal interrupt reaches the child once and allows cleanup to finish.""" + script = tmp_path / "sigint_cleanup.py" + marker = tmp_path / "cleanup-count" + script.write_text(_SIGINT_CLEANUP_SCRIPT) + with subprocess.Popen( + [sys.executable, str(script), str(marker)], + stdout=subprocess.PIPE, + env={**os.environ, "REFLEX_LOG_JSON": "true"}, + start_new_session=True, + ) as proc: + assert proc.stdout is not None + assert json.loads(proc.stdout.readline())["message"] == "ready" + killpg = getattr(os, "killpg", None) + assert callable(killpg) + killpg(proc.pid, signal.SIGINT) + assert proc.wait(timeout=10) == 0 + + assert marker.read_text() == "1" + + def test_output_pump_survives_a_gone_consumer(): """The reader keeps draining the child when its consumer closed the stream.""" child_read, child_write = os.pipe() From 4786c04de93772c963179532f306889a04eb5124 Mon Sep 17 00:00:00 2001 From: Harsh Thakare Date: Wed, 14 Oct 2026 16:13:47 +0530 Subject: [PATCH 06/11] fix: forward supervised process shutdown signals --- .../reflex-base/src/reflex_base/utils/log.py | 90 ++++++++++++------- tests/units/reflex_base/utils/test_log.py | 76 ++++++++++++++-- 2 files changed, 131 insertions(+), 35 deletions(-) diff --git a/packages/reflex-base/src/reflex_base/utils/log.py b/packages/reflex-base/src/reflex_base/utils/log.py index b86fde6121a..7db00ea629c 100644 --- a/packages/reflex-base/src/reflex_base/utils/log.py +++ b/packages/reflex-base/src/reflex_base/utils/log.py @@ -503,36 +503,66 @@ def supervise_output(args: list[str]) -> int: _SUPERVISED_ENV_VAR: str(os.getpid()), "PYTHONIOENCODING": f"utf-8:{errors}" if errors else "utf-8", } - # Raw pipes that only the readers close: closing a pipe while a reader is - # blocked on it waits for that read on Windows. - out_read, out_write = os.pipe() - err_read, err_write = os.pipe() - proc = subprocess.Popen( - args, - stdout=out_write, - stderr=err_write, - env=env, - start_new_session=os.name == "posix", - ) - os.close(out_write) - os.close(err_write) - pumps = [ - _OutputPump(out_read, 1, "info", "stdout"), - _OutputPump(err_read, 2, "warning", "stderr"), - ] - for pump in pumps: - pump.start() - signal.signal(signal.SIGTERM, lambda *_: proc.terminate()) - while True: - try: - returncode = proc.wait() - break - except KeyboardInterrupt: - # Keep terminal SIGINT from reaching the child before we forward it. - if os.name == "posix" and proc.poll() is None: - with contextlib.suppress(ProcessLookupError): - os.killpg(proc.pid, signal.SIGINT) - continue + proc: subprocess.Popen[bytes] | None = None + pending_signals: list[int] = [] + + def forward_signal(signum: int, _frame: object) -> None: + """Forward shutdown signals, queueing them until the child exists.""" + if proc is None: + pending_signals.append(signum) + elif proc.poll() is None: + with contextlib.suppress(ProcessLookupError): + killpg = getattr(os, "killpg", None) + if callable(killpg): + killpg(proc.pid, signum) + + previous_handlers: list[tuple[int, signal.Handlers]] = [] + if os.name == "posix": + forwarded_signals = [signal.SIGINT, signal.SIGTERM] + sighup = getattr(signal, "SIGHUP", None) + if sighup is not None: + forwarded_signals.append(sighup) + previous_handlers.extend( + (signum, signal.signal(signum, forward_signal)) + for signum in forwarded_signals + ) + + try: + # Raw pipes that only the readers close: closing a pipe while a reader + # is blocked on it waits for that read on Windows. + out_read, out_write = os.pipe() + err_read, err_write = os.pipe() + proc = subprocess.Popen( + args, + stdout=out_write, + stderr=err_write, + env=env, + start_new_session=os.name == "posix", + ) + os.close(out_write) + os.close(err_write) + pumps = [ + _OutputPump(out_read, 1, "info", "stdout"), + _OutputPump(err_read, 2, "warning", "stderr"), + ] + for pump in pumps: + pump.start() + if os.name != "posix": + signal.signal(signal.SIGTERM, lambda *_: proc.terminate()) + for signum in pending_signals: + forward_signal(signum, None) + pending_signals.clear() + while True: + try: + returncode = proc.wait() + break + except KeyboardInterrupt: + # On Windows the child shares the console process group. + if os.name == "posix": + raise + finally: + for signum, handler in previous_handlers: + signal.signal(signum, handler) wall_deadline = time.monotonic() + _DRAIN_WALL_SECONDS snapshots = [pump.writing_snapshot() for pump in pumps] for pump, (exited_at, writing_at_exit) in zip(pumps, snapshots, strict=True): diff --git a/tests/units/reflex_base/utils/test_log.py b/tests/units/reflex_base/utils/test_log.py index bc214f7211f..af16e791ff5 100644 --- a/tests/units/reflex_base/utils/test_log.py +++ b/tests/units/reflex_base/utils/test_log.py @@ -1184,6 +1184,7 @@ def test_supervise_output_forwards_pid_targeted_sigint(tmp_path): _SIGINT_CLEANUP_SCRIPT = """ +import os import signal import sys import time @@ -1192,20 +1193,39 @@ def test_supervise_output_forwards_pid_targeted_sigint(tmp_path): from reflex_base.utils import log if not log.is_output_supervised(): - sys.exit(log.supervise_output([sys.executable, __file__, sys.argv[1]])) + mode, marker, ready = sys.argv[1:] + if mode == "setup": + real_popen = log.subprocess.Popen + + def spawn_and_interrupt(*args, **kwargs): + child = real_popen(*args, **kwargs) + deadline = time.monotonic() + 10 + while not Path(ready).exists() and time.monotonic() < deadline: + time.sleep(0.01) + assert Path(ready).exists() + os.kill(os.getpid(), signal.SIGINT) + return child + + log.subprocess.Popen = spawn_and_interrupt + sys.exit( + log.supervise_output([sys.executable, __file__, mode, marker, ready]) + ) signal_count = 0 +mode, marker, ready = sys.argv[1:] +signal_number = getattr(signal, "SIGHUP") if mode == "hangup" else signal.SIGINT -def handle_sigint(*_): +def handle_shutdown(*_): global signal_count signal_count += 1 -signal.signal(signal.SIGINT, handle_sigint) +signal.signal(signal_number, handle_shutdown) +Path(ready).write_text("ready") print("ready", flush=True) while signal_count == 0: time.sleep(0.01) time.sleep(0.25) -Path(sys.argv[1]).write_text(str(signal_count)) +Path(marker).write_text(str(signal_count)) """ @@ -1214,9 +1234,10 @@ def test_supervise_output_does_not_double_forward_terminal_sigint(tmp_path): """A terminal interrupt reaches the child once and allows cleanup to finish.""" script = tmp_path / "sigint_cleanup.py" marker = tmp_path / "cleanup-count" + ready = tmp_path / "ready" script.write_text(_SIGINT_CLEANUP_SCRIPT) with subprocess.Popen( - [sys.executable, str(script), str(marker)], + [sys.executable, str(script), "terminal", str(marker), str(ready)], stdout=subprocess.PIPE, env={**os.environ, "REFLEX_LOG_JSON": "true"}, start_new_session=True, @@ -1231,6 +1252,51 @@ def test_supervise_output_does_not_double_forward_terminal_sigint(tmp_path): assert marker.read_text() == "1" +@pytest.mark.skipif(sys.platform == "win32", reason="SIGINT is POSIX-only") +def test_supervise_output_forwards_sigint_during_setup(tmp_path): + """A SIGINT between spawn and wait is queued until the child can be signaled.""" + script = tmp_path / "sigint_cleanup.py" + marker = tmp_path / "cleanup-count" + ready = tmp_path / "ready" + script.write_text(_SIGINT_CLEANUP_SCRIPT) + with subprocess.Popen( + [sys.executable, str(script), "setup", str(marker), str(ready)], + stdout=subprocess.PIPE, + env={**os.environ, "REFLEX_LOG_JSON": "true"}, + start_new_session=True, + ) as proc: + assert proc.stdout is not None + assert json.loads(proc.stdout.readline())["message"] == "ready" + assert proc.wait(timeout=10) == 0 + + assert marker.read_text() == "1" + + +@pytest.mark.skipif(sys.platform == "win32", reason="SIGHUP is POSIX-only") +def test_supervise_output_forwards_terminal_sighup(tmp_path): + """A terminal hangup reaches the child and waits for its cleanup.""" + script = tmp_path / "sigint_cleanup.py" + marker = tmp_path / "cleanup-count" + ready = tmp_path / "ready" + script.write_text(_SIGINT_CLEANUP_SCRIPT) + with subprocess.Popen( + [sys.executable, str(script), "hangup", str(marker), str(ready)], + stdout=subprocess.PIPE, + env={**os.environ, "REFLEX_LOG_JSON": "true"}, + start_new_session=True, + ) as proc: + assert proc.stdout is not None + assert json.loads(proc.stdout.readline())["message"] == "ready" + killpg = getattr(os, "killpg", None) + sighup = getattr(signal, "SIGHUP", None) + assert callable(killpg) + assert sighup is not None + killpg(proc.pid, sighup) + assert proc.wait(timeout=10) == 0 + + assert marker.read_text() == "1" + + def test_output_pump_survives_a_gone_consumer(): """The reader keeps draining the child when its consumer closed the stream.""" child_read, child_write = os.pipe() From 14a2b77ae0d1b5b702595ebab785736bf9f856ff Mon Sep 17 00:00:00 2001 From: Harsh Thakare Date: Wed, 14 Oct 2026 16:23:38 +0530 Subject: [PATCH 07/11] fix: include greenlet in legacy playground check --- .github/workflows/examples.yml | 2 +- packages/reflex-base/src/reflex_base/utils/log.py | 13 +++++++++---- 2 files changed, 10 insertions(+), 5 deletions(-) diff --git a/.github/workflows/examples.yml b/.github/workflows/examples.yml index c34ec5b014a..d1ed9b11493 100644 --- a/.github/workflows/examples.yml +++ b/.github/workflows/examples.yml @@ -82,7 +82,7 @@ jobs: - name: Install reflex 0.8.23 into a fresh venv run: | uv venv --python 3.12 "$RUNNER_TEMP/reflex-0.8.23" - uv pip install --python "$RUNNER_TEMP/reflex-0.8.23/bin/python" "reflex==0.8.23" "plotly>=6.0,<7.0" + uv pip install --python "$RUNNER_TEMP/reflex-0.8.23/bin/python" "reflex==0.8.23" "plotly>=6.0,<7.0" greenlet echo "$RUNNER_TEMP/reflex-0.8.23/bin" >> "$GITHUB_PATH" - name: Show the reflex under test working-directory: examples/playground diff --git a/packages/reflex-base/src/reflex_base/utils/log.py b/packages/reflex-base/src/reflex_base/utils/log.py index 7db00ea629c..1a7a85ffd10 100644 --- a/packages/reflex-base/src/reflex_base/utils/log.py +++ b/packages/reflex-base/src/reflex_base/utils/log.py @@ -33,10 +33,10 @@ import sys import threading import time -from collections.abc import Generator +from collections.abc import Callable, Generator from pathlib import Path from types import FrameType, ModuleType -from typing import TYPE_CHECKING, cast +from typing import TYPE_CHECKING, Any, cast from rich.console import Console from rich.errors import MarkupError @@ -516,14 +516,19 @@ def forward_signal(signum: int, _frame: object) -> None: if callable(killpg): killpg(proc.pid, signum) - previous_handlers: list[tuple[int, signal.Handlers]] = [] + previous_handlers: list[ + tuple[ + int, + int | signal.Handlers | Callable[[int, FrameType | None], Any] | None, + ] + ] = [] if os.name == "posix": forwarded_signals = [signal.SIGINT, signal.SIGTERM] sighup = getattr(signal, "SIGHUP", None) if sighup is not None: forwarded_signals.append(sighup) previous_handlers.extend( - (signum, signal.signal(signum, forward_signal)) + (int(signum), signal.signal(signum, forward_signal)) for signum in forwarded_signals ) From b78bd1a7d1a181d5c711352bb2cf1679d799b140 Mon Sep 17 00:00:00 2001 From: Harsh Thakare Date: Thu, 15 Oct 2026 09:43:00 +0530 Subject: [PATCH 08/11] fix: forward SIGQUIT to supervised process --- .../reflex-base/src/reflex_base/utils/log.py | 7 +++-- tests/units/reflex_base/utils/test_log.py | 28 ++++++++++++++++++- 2 files changed, 31 insertions(+), 4 deletions(-) diff --git a/packages/reflex-base/src/reflex_base/utils/log.py b/packages/reflex-base/src/reflex_base/utils/log.py index 1a7a85ffd10..bf0b38694e9 100644 --- a/packages/reflex-base/src/reflex_base/utils/log.py +++ b/packages/reflex-base/src/reflex_base/utils/log.py @@ -524,9 +524,10 @@ def forward_signal(signum: int, _frame: object) -> None: ] = [] if os.name == "posix": forwarded_signals = [signal.SIGINT, signal.SIGTERM] - sighup = getattr(signal, "SIGHUP", None) - if sighup is not None: - forwarded_signals.append(sighup) + for signal_name in ("SIGHUP", "SIGQUIT"): + signum = getattr(signal, signal_name, None) + if signum is not None: + forwarded_signals.append(signum) previous_handlers.extend( (int(signum), signal.signal(signum, forward_signal)) for signum in forwarded_signals diff --git a/tests/units/reflex_base/utils/test_log.py b/tests/units/reflex_base/utils/test_log.py index af16e791ff5..5095fb4b487 100644 --- a/tests/units/reflex_base/utils/test_log.py +++ b/tests/units/reflex_base/utils/test_log.py @@ -1213,7 +1213,8 @@ def spawn_and_interrupt(*args, **kwargs): signal_count = 0 mode, marker, ready = sys.argv[1:] -signal_number = getattr(signal, "SIGHUP") if mode == "hangup" else signal.SIGINT +signal_name = {"hangup": "SIGHUP", "quit": "SIGQUIT"}.get(mode, "SIGINT") +signal_number = getattr(signal, signal_name, signal.SIGINT) def handle_shutdown(*_): global signal_count @@ -1297,6 +1298,31 @@ def test_supervise_output_forwards_terminal_sighup(tmp_path): assert marker.read_text() == "1" +@pytest.mark.skipif(sys.platform == "win32", reason="SIGQUIT is POSIX-only") +def test_supervise_output_forwards_terminal_sigquit(tmp_path): + """A terminal quit reaches the child and waits for its cleanup.""" + script = tmp_path / "sigint_cleanup.py" + marker = tmp_path / "cleanup-count" + ready = tmp_path / "ready" + script.write_text(_SIGINT_CLEANUP_SCRIPT) + with subprocess.Popen( + [sys.executable, str(script), "quit", str(marker), str(ready)], + stdout=subprocess.PIPE, + env={**os.environ, "REFLEX_LOG_JSON": "true"}, + start_new_session=True, + ) as proc: + assert proc.stdout is not None + assert json.loads(proc.stdout.readline())["message"] == "ready" + killpg = getattr(os, "killpg", None) + sigquit = getattr(signal, "SIGQUIT", None) + assert callable(killpg) + assert sigquit is not None + killpg(proc.pid, sigquit) + assert proc.wait(timeout=10) == 0 + + assert marker.read_text() == "1" + + def test_output_pump_survives_a_gone_consumer(): """The reader keeps draining the child when its consumer closed the stream.""" child_read, child_write = os.pipe() From cc7d2423419a8ffd8897e300348d8bdf8ba789cc Mon Sep 17 00:00:00 2001 From: Harsh Thakare Date: Thu, 22 Oct 2026 11:31:26 +0530 Subject: [PATCH 09/11] fix: preserve existing handlers in JSON supervisor --- .../reflex-base/src/reflex_base/utils/log.py | 29 +++++++----- tests/units/reflex_base/utils/test_log.py | 45 +++++++++++++++++++ 2 files changed, 62 insertions(+), 12 deletions(-) diff --git a/packages/reflex-base/src/reflex_base/utils/log.py b/packages/reflex-base/src/reflex_base/utils/log.py index bf0b38694e9..29e2bf81a53 100644 --- a/packages/reflex-base/src/reflex_base/utils/log.py +++ b/packages/reflex-base/src/reflex_base/utils/log.py @@ -506,7 +506,12 @@ def supervise_output(args: list[str]) -> int: proc: subprocess.Popen[bytes] | None = None pending_signals: list[int] = [] - def forward_signal(signum: int, _frame: object) -> None: + previous_handlers: dict[ + int, + int | signal.Handlers | Callable[[int, FrameType | None], Any] | None, + ] = {} + + def forward_signal(signum: int, frame: FrameType | None) -> None: """Forward shutdown signals, queueing them until the child exists.""" if proc is None: pending_signals.append(signum) @@ -515,23 +520,23 @@ def forward_signal(signum: int, _frame: object) -> None: killpg = getattr(os, "killpg", None) if callable(killpg): killpg(proc.pid, signum) + previous_handler = previous_handlers.get(signum) + # The default SIGINT handler raises KeyboardInterrupt before the child exits. + if ( + callable(previous_handler) + and previous_handler is not signal.default_int_handler + ): + previous_handler(signum, frame) - previous_handlers: list[ - tuple[ - int, - int | signal.Handlers | Callable[[int, FrameType | None], Any] | None, - ] - ] = [] if os.name == "posix": forwarded_signals = [signal.SIGINT, signal.SIGTERM] for signal_name in ("SIGHUP", "SIGQUIT"): signum = getattr(signal, signal_name, None) if signum is not None: forwarded_signals.append(signum) - previous_handlers.extend( - (int(signum), signal.signal(signum, forward_signal)) - for signum in forwarded_signals - ) + for signum in forwarded_signals: + previous_handlers[int(signum)] = signal.getsignal(signum) + signal.signal(signum, forward_signal) try: # Raw pipes that only the readers close: closing a pipe while a reader @@ -567,7 +572,7 @@ def forward_signal(signum: int, _frame: object) -> None: if os.name == "posix": raise finally: - for signum, handler in previous_handlers: + for signum, handler in previous_handlers.items(): signal.signal(signum, handler) wall_deadline = time.monotonic() + _DRAIN_WALL_SECONDS snapshots = [pump.writing_snapshot() for pump in pumps] diff --git a/tests/units/reflex_base/utils/test_log.py b/tests/units/reflex_base/utils/test_log.py index 5095fb4b487..34353878b50 100644 --- a/tests/units/reflex_base/utils/test_log.py +++ b/tests/units/reflex_base/utils/test_log.py @@ -1167,6 +1167,51 @@ def test_supervise_output_forwards_sigterm(tmp_path): assert proc.wait(timeout=10) == 128 + signal.SIGTERM +@pytest.mark.skipif(sys.platform == "win32", reason="SIGTERM is POSIX-only") +def test_supervise_output_calls_and_restores_previous_signal_handler(tmp_path): + """A custom signal handler still runs and is restored after supervision.""" + script = tmp_path / "sigterm_previous_handler.py" + marker = tmp_path / "previous-handler-called" + script.write_text( + """ +import signal +import sys +import time +from pathlib import Path + +from reflex_base.utils import log + +if not log.is_output_supervised(): + marker = Path(sys.argv[1]) + + def previous_handler(signum, frame): + marker.write_text(str(signum)) + + signal.signal(signal.SIGTERM, previous_handler) + returncode = log.supervise_output([sys.executable, __file__, str(marker)]) + marker.write_text( + f"{marker.read_text() if marker.exists() else 'not-called'}" + f"\\nrestored={signal.getsignal(signal.SIGTERM) is previous_handler}" + ) + sys.exit(returncode) + +print("ready", flush=True) +time.sleep(30) +""" + ) + with subprocess.Popen( + [sys.executable, str(script), str(marker)], + stdout=subprocess.PIPE, + env={**os.environ, "REFLEX_LOG_JSON": "true"}, + ) as proc: + assert proc.stdout is not None + assert json.loads(proc.stdout.readline())["message"] == "ready" + proc.send_signal(signal.SIGTERM) + assert proc.wait(timeout=10) == 128 + signal.SIGTERM + + assert marker.read_text() == f"{signal.SIGTERM}\nrestored=True" + + @pytest.mark.skipif(sys.platform == "win32", reason="SIGINT is POSIX-only") def test_supervise_output_forwards_pid_targeted_sigint(tmp_path): """A PID-targeted interrupt reaches the supervised command.""" From b4b5261afb3264f93d6ce15b422575a321a0a758 Mon Sep 17 00:00:00 2001 From: Harsh Thakare Date: Thu, 22 Oct 2026 11:33:32 +0530 Subject: [PATCH 10/11] test: ensure queued signals call prior handler once --- .../reflex-base/src/reflex_base/utils/log.py | 17 +++++++++++------ tests/units/reflex_base/utils/test_log.py | 8 ++++++++ 2 files changed, 19 insertions(+), 6 deletions(-) diff --git a/packages/reflex-base/src/reflex_base/utils/log.py b/packages/reflex-base/src/reflex_base/utils/log.py index 29e2bf81a53..18791ab0374 100644 --- a/packages/reflex-base/src/reflex_base/utils/log.py +++ b/packages/reflex-base/src/reflex_base/utils/log.py @@ -511,15 +511,20 @@ def supervise_output(args: list[str]) -> int: int | signal.Handlers | Callable[[int, FrameType | None], Any] | None, ] = {} - def forward_signal(signum: int, frame: FrameType | None) -> None: - """Forward shutdown signals, queueing them until the child exists.""" - if proc is None: - pending_signals.append(signum) - elif proc.poll() is None: + def forward_to_child(signum: int) -> None: + """Forward one signal to the running child process group.""" + if proc is not None and proc.poll() is None: with contextlib.suppress(ProcessLookupError): killpg = getattr(os, "killpg", None) if callable(killpg): killpg(proc.pid, signum) + + def forward_signal(signum: int, frame: FrameType | None) -> None: + """Forward shutdown signals, queueing them until the child exists.""" + if proc is None: + pending_signals.append(signum) + else: + forward_to_child(signum) previous_handler = previous_handlers.get(signum) # The default SIGINT handler raises KeyboardInterrupt before the child exits. if ( @@ -561,7 +566,7 @@ def forward_signal(signum: int, frame: FrameType | None) -> None: if os.name != "posix": signal.signal(signal.SIGTERM, lambda *_: proc.terminate()) for signum in pending_signals: - forward_signal(signum, None) + forward_to_child(signum) pending_signals.clear() while True: try: diff --git a/tests/units/reflex_base/utils/test_log.py b/tests/units/reflex_base/utils/test_log.py index 34353878b50..053cd6f7a63 100644 --- a/tests/units/reflex_base/utils/test_log.py +++ b/tests/units/reflex_base/utils/test_log.py @@ -1239,6 +1239,12 @@ def test_supervise_output_forwards_pid_targeted_sigint(tmp_path): if not log.is_output_supervised(): mode, marker, ready = sys.argv[1:] + + def previous_handler(signum, frame): + with Path(f"{marker}.handler").open("a") as handler_calls: + handler_calls.write("x") + + signal.signal(signal.SIGINT, previous_handler) if mode == "setup": real_popen = log.subprocess.Popen @@ -1296,6 +1302,7 @@ def test_supervise_output_does_not_double_forward_terminal_sigint(tmp_path): assert proc.wait(timeout=10) == 0 assert marker.read_text() == "1" + assert Path(f"{marker}.handler").read_text() == "x" @pytest.mark.skipif(sys.platform == "win32", reason="SIGINT is POSIX-only") @@ -1316,6 +1323,7 @@ def test_supervise_output_forwards_sigint_during_setup(tmp_path): assert proc.wait(timeout=10) == 0 assert marker.read_text() == "1" + assert Path(f"{marker}.handler").read_text() == "x" @pytest.mark.skipif(sys.platform == "win32", reason="SIGHUP is POSIX-only") From 450a2dea06f82b941025340f07f2ec5ea027cbae Mon Sep 17 00:00:00 2001 From: Farhan Date: Thu, 22 Oct 2026 22:14:28 +0500 Subject: [PATCH 11/11] fix: forward PID-targeted SIGINT without a new session Keep the child in the supervisor's process group so a group SIGKILL still stops the whole app, and forward SIGINT from the KeyboardInterrupt branch instead of replacing signal handlers. The app shuts down cleanly when a terminal Ctrl-C reaches it twice. Drop SIGHUP/SIGQUIT forwarding: granian reloads its workers on SIGHUP, so forwarding it does not stop the app. --- .../reflex-base/src/reflex_base/utils/log.py | 103 ++------ tests/units/reflex_base/utils/test_log.py | 230 +----------------- 2 files changed, 38 insertions(+), 295 deletions(-) diff --git a/packages/reflex-base/src/reflex_base/utils/log.py b/packages/reflex-base/src/reflex_base/utils/log.py index 18791ab0374..725313fe889 100644 --- a/packages/reflex-base/src/reflex_base/utils/log.py +++ b/packages/reflex-base/src/reflex_base/utils/log.py @@ -33,10 +33,10 @@ import sys import threading import time -from collections.abc import Callable, Generator +from collections.abc import Generator from pathlib import Path from types import FrameType, ModuleType -from typing import TYPE_CHECKING, Any, cast +from typing import TYPE_CHECKING, cast from rich.console import Console from rich.errors import MarkupError @@ -503,82 +503,29 @@ def supervise_output(args: list[str]) -> int: _SUPERVISED_ENV_VAR: str(os.getpid()), "PYTHONIOENCODING": f"utf-8:{errors}" if errors else "utf-8", } - proc: subprocess.Popen[bytes] | None = None - pending_signals: list[int] = [] - - previous_handlers: dict[ - int, - int | signal.Handlers | Callable[[int, FrameType | None], Any] | None, - ] = {} - - def forward_to_child(signum: int) -> None: - """Forward one signal to the running child process group.""" - if proc is not None and proc.poll() is None: - with contextlib.suppress(ProcessLookupError): - killpg = getattr(os, "killpg", None) - if callable(killpg): - killpg(proc.pid, signum) - - def forward_signal(signum: int, frame: FrameType | None) -> None: - """Forward shutdown signals, queueing them until the child exists.""" - if proc is None: - pending_signals.append(signum) - else: - forward_to_child(signum) - previous_handler = previous_handlers.get(signum) - # The default SIGINT handler raises KeyboardInterrupt before the child exits. - if ( - callable(previous_handler) - and previous_handler is not signal.default_int_handler - ): - previous_handler(signum, frame) - - if os.name == "posix": - forwarded_signals = [signal.SIGINT, signal.SIGTERM] - for signal_name in ("SIGHUP", "SIGQUIT"): - signum = getattr(signal, signal_name, None) - if signum is not None: - forwarded_signals.append(signum) - for signum in forwarded_signals: - previous_handlers[int(signum)] = signal.getsignal(signum) - signal.signal(signum, forward_signal) - - try: - # Raw pipes that only the readers close: closing a pipe while a reader - # is blocked on it waits for that read on Windows. - out_read, out_write = os.pipe() - err_read, err_write = os.pipe() - proc = subprocess.Popen( - args, - stdout=out_write, - stderr=err_write, - env=env, - start_new_session=os.name == "posix", - ) - os.close(out_write) - os.close(err_write) - pumps = [ - _OutputPump(out_read, 1, "info", "stdout"), - _OutputPump(err_read, 2, "warning", "stderr"), - ] - for pump in pumps: - pump.start() - if os.name != "posix": - signal.signal(signal.SIGTERM, lambda *_: proc.terminate()) - for signum in pending_signals: - forward_to_child(signum) - pending_signals.clear() - while True: - try: - returncode = proc.wait() - break - except KeyboardInterrupt: - # On Windows the child shares the console process group. - if os.name == "posix": - raise - finally: - for signum, handler in previous_handlers.items(): - signal.signal(signum, handler) + # Raw pipes that only the readers close: closing a pipe while a reader is + # blocked on it waits for that read on Windows. + out_read, out_write = os.pipe() + err_read, err_write = os.pipe() + proc = subprocess.Popen(args, stdout=out_write, stderr=err_write, env=env) + os.close(out_write) + os.close(err_write) + pumps = [ + _OutputPump(out_read, 1, "info", "stdout"), + _OutputPump(err_read, 2, "warning", "stderr"), + ] + for pump in pumps: + pump.start() + signal.signal(signal.SIGTERM, lambda *_: proc.terminate()) + while True: + try: + returncode = proc.wait() + break + except KeyboardInterrupt: + # A terminal Ctrl-C reaches the child too, a SIGINT sent only to + # this process does not. The child handles a repeated interrupt. + if sys.platform != "win32": + proc.send_signal(signal.SIGINT) wall_deadline = time.monotonic() + _DRAIN_WALL_SECONDS snapshots = [pump.writing_snapshot() for pump in pumps] for pump, (exited_at, writing_at_exit) in zip(pumps, snapshots, strict=True): diff --git a/tests/units/reflex_base/utils/test_log.py b/tests/units/reflex_base/utils/test_log.py index 053cd6f7a63..dfdf620957e 100644 --- a/tests/units/reflex_base/utils/test_log.py +++ b/tests/units/reflex_base/utils/test_log.py @@ -1138,7 +1138,7 @@ def test_supervise_output_does_not_wait_for_lingering_descendants(tmp_path): assert elapsed < 10 -_SIGTERM_SCRIPT = """ +_SLEEP_SCRIPT = """ import sys import time @@ -1151,72 +1151,16 @@ def test_supervise_output_does_not_wait_for_lingering_descendants(tmp_path): """ -@pytest.mark.skipif(sys.platform == "win32", reason="SIGTERM is POSIX-only") -def test_supervise_output_forwards_sigterm(tmp_path): - """Terminating the supervisor terminates the command and reports the signal.""" - script = tmp_path / "sigterm.py" - script.write_text(_SIGTERM_SCRIPT) - with subprocess.Popen( - [sys.executable, str(script)], - stdout=subprocess.PIPE, - env={**os.environ, "REFLEX_LOG_JSON": "true"}, - ) as proc: - assert proc.stdout is not None - assert json.loads(proc.stdout.readline())["message"] == "ready" - proc.send_signal(signal.SIGTERM) - assert proc.wait(timeout=10) == 128 + signal.SIGTERM - - -@pytest.mark.skipif(sys.platform == "win32", reason="SIGTERM is POSIX-only") -def test_supervise_output_calls_and_restores_previous_signal_handler(tmp_path): - """A custom signal handler still runs and is restored after supervision.""" - script = tmp_path / "sigterm_previous_handler.py" - marker = tmp_path / "previous-handler-called" - script.write_text( - """ -import signal -import sys -import time -from pathlib import Path - -from reflex_base.utils import log - -if not log.is_output_supervised(): - marker = Path(sys.argv[1]) - - def previous_handler(signum, frame): - marker.write_text(str(signum)) - - signal.signal(signal.SIGTERM, previous_handler) - returncode = log.supervise_output([sys.executable, __file__, str(marker)]) - marker.write_text( - f"{marker.read_text() if marker.exists() else 'not-called'}" - f"\\nrestored={signal.getsignal(signal.SIGTERM) is previous_handler}" - ) - sys.exit(returncode) - -print("ready", flush=True) -time.sleep(30) -""" - ) - with subprocess.Popen( - [sys.executable, str(script), str(marker)], - stdout=subprocess.PIPE, - env={**os.environ, "REFLEX_LOG_JSON": "true"}, - ) as proc: - assert proc.stdout is not None - assert json.loads(proc.stdout.readline())["message"] == "ready" - proc.send_signal(signal.SIGTERM) - assert proc.wait(timeout=10) == 128 + signal.SIGTERM - - assert marker.read_text() == f"{signal.SIGTERM}\nrestored=True" - - -@pytest.mark.skipif(sys.platform == "win32", reason="SIGINT is POSIX-only") -def test_supervise_output_forwards_pid_targeted_sigint(tmp_path): - """A PID-targeted interrupt reaches the supervised command.""" - script = tmp_path / "sigint.py" - script.write_text(_SIGTERM_SCRIPT) +@pytest.mark.skipif( + sys.platform == "win32", reason="PID-targeted signals are POSIX-only" +) +@pytest.mark.parametrize( + "signum", [signal.SIGTERM, signal.SIGINT], ids=["sigterm", "sigint"] +) +def test_supervise_output_forwards_shutdown_signal(tmp_path, signum): + """A signal sent to the supervisor's PID stops the command and is reported.""" + script = tmp_path / "sleep.py" + script.write_text(_SLEEP_SCRIPT) with subprocess.Popen( [sys.executable, str(script)], stdout=subprocess.PIPE, @@ -1224,156 +1168,8 @@ def test_supervise_output_forwards_pid_targeted_sigint(tmp_path): ) as proc: assert proc.stdout is not None assert json.loads(proc.stdout.readline())["message"] == "ready" - proc.send_signal(signal.SIGINT) - assert proc.wait(timeout=10) == 128 + signal.SIGINT - - -_SIGINT_CLEANUP_SCRIPT = """ -import os -import signal -import sys -import time -from pathlib import Path - -from reflex_base.utils import log - -if not log.is_output_supervised(): - mode, marker, ready = sys.argv[1:] - - def previous_handler(signum, frame): - with Path(f"{marker}.handler").open("a") as handler_calls: - handler_calls.write("x") - - signal.signal(signal.SIGINT, previous_handler) - if mode == "setup": - real_popen = log.subprocess.Popen - - def spawn_and_interrupt(*args, **kwargs): - child = real_popen(*args, **kwargs) - deadline = time.monotonic() + 10 - while not Path(ready).exists() and time.monotonic() < deadline: - time.sleep(0.01) - assert Path(ready).exists() - os.kill(os.getpid(), signal.SIGINT) - return child - - log.subprocess.Popen = spawn_and_interrupt - sys.exit( - log.supervise_output([sys.executable, __file__, mode, marker, ready]) - ) - -signal_count = 0 -mode, marker, ready = sys.argv[1:] -signal_name = {"hangup": "SIGHUP", "quit": "SIGQUIT"}.get(mode, "SIGINT") -signal_number = getattr(signal, signal_name, signal.SIGINT) - -def handle_shutdown(*_): - global signal_count - signal_count += 1 - -signal.signal(signal_number, handle_shutdown) -Path(ready).write_text("ready") -print("ready", flush=True) -while signal_count == 0: - time.sleep(0.01) -time.sleep(0.25) -Path(marker).write_text(str(signal_count)) -""" - - -@pytest.mark.skipif(sys.platform == "win32", reason="SIGINT is POSIX-only") -def test_supervise_output_does_not_double_forward_terminal_sigint(tmp_path): - """A terminal interrupt reaches the child once and allows cleanup to finish.""" - script = tmp_path / "sigint_cleanup.py" - marker = tmp_path / "cleanup-count" - ready = tmp_path / "ready" - script.write_text(_SIGINT_CLEANUP_SCRIPT) - with subprocess.Popen( - [sys.executable, str(script), "terminal", str(marker), str(ready)], - stdout=subprocess.PIPE, - env={**os.environ, "REFLEX_LOG_JSON": "true"}, - start_new_session=True, - ) as proc: - assert proc.stdout is not None - assert json.loads(proc.stdout.readline())["message"] == "ready" - killpg = getattr(os, "killpg", None) - assert callable(killpg) - killpg(proc.pid, signal.SIGINT) - assert proc.wait(timeout=10) == 0 - - assert marker.read_text() == "1" - assert Path(f"{marker}.handler").read_text() == "x" - - -@pytest.mark.skipif(sys.platform == "win32", reason="SIGINT is POSIX-only") -def test_supervise_output_forwards_sigint_during_setup(tmp_path): - """A SIGINT between spawn and wait is queued until the child can be signaled.""" - script = tmp_path / "sigint_cleanup.py" - marker = tmp_path / "cleanup-count" - ready = tmp_path / "ready" - script.write_text(_SIGINT_CLEANUP_SCRIPT) - with subprocess.Popen( - [sys.executable, str(script), "setup", str(marker), str(ready)], - stdout=subprocess.PIPE, - env={**os.environ, "REFLEX_LOG_JSON": "true"}, - start_new_session=True, - ) as proc: - assert proc.stdout is not None - assert json.loads(proc.stdout.readline())["message"] == "ready" - assert proc.wait(timeout=10) == 0 - - assert marker.read_text() == "1" - assert Path(f"{marker}.handler").read_text() == "x" - - -@pytest.mark.skipif(sys.platform == "win32", reason="SIGHUP is POSIX-only") -def test_supervise_output_forwards_terminal_sighup(tmp_path): - """A terminal hangup reaches the child and waits for its cleanup.""" - script = tmp_path / "sigint_cleanup.py" - marker = tmp_path / "cleanup-count" - ready = tmp_path / "ready" - script.write_text(_SIGINT_CLEANUP_SCRIPT) - with subprocess.Popen( - [sys.executable, str(script), "hangup", str(marker), str(ready)], - stdout=subprocess.PIPE, - env={**os.environ, "REFLEX_LOG_JSON": "true"}, - start_new_session=True, - ) as proc: - assert proc.stdout is not None - assert json.loads(proc.stdout.readline())["message"] == "ready" - killpg = getattr(os, "killpg", None) - sighup = getattr(signal, "SIGHUP", None) - assert callable(killpg) - assert sighup is not None - killpg(proc.pid, sighup) - assert proc.wait(timeout=10) == 0 - - assert marker.read_text() == "1" - - -@pytest.mark.skipif(sys.platform == "win32", reason="SIGQUIT is POSIX-only") -def test_supervise_output_forwards_terminal_sigquit(tmp_path): - """A terminal quit reaches the child and waits for its cleanup.""" - script = tmp_path / "sigint_cleanup.py" - marker = tmp_path / "cleanup-count" - ready = tmp_path / "ready" - script.write_text(_SIGINT_CLEANUP_SCRIPT) - with subprocess.Popen( - [sys.executable, str(script), "quit", str(marker), str(ready)], - stdout=subprocess.PIPE, - env={**os.environ, "REFLEX_LOG_JSON": "true"}, - start_new_session=True, - ) as proc: - assert proc.stdout is not None - assert json.loads(proc.stdout.readline())["message"] == "ready" - killpg = getattr(os, "killpg", None) - sigquit = getattr(signal, "SIGQUIT", None) - assert callable(killpg) - assert sigquit is not None - killpg(proc.pid, sigquit) - assert proc.wait(timeout=10) == 0 - - assert marker.read_text() == "1" + proc.send_signal(signum) + assert proc.wait(timeout=10) == 128 + signum def test_output_pump_survives_a_gone_consumer():