Forward POSIX shutdown signals to supervised processes in JSON mode

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.

packages/reflex-base/src/reflex_base/utils/log.py

@@ -522,8 +522,10 @@ 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.
-            continue
+            # 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):

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,20 +1151,25 @@ 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)
+@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,
         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
+        proc.send_signal(signum)
+        assert proc.wait(timeout=10) == 128 + signum
 
 
 def test_output_pump_survives_a_gone_consumer():