fix(testgen): kill the whole subprocess tree on tg start/install shutdown

`proc.terminate()` on Windows calls `TerminateProcess` and only kills the
parent — pixeltable-pgserver's `postgres.exe` children get orphaned. The
symptom: `tg delete` fails with "Could not remove TestGen data directory"
because postgres holds the data files open.

New `stop_app_tree(proc, timeout=...)` walks the descendant tree:
- Windows: shells out to `taskkill /F /T /PID` to walk the WMI process tree.
- POSIX: sends SIGTERM to the process group (parent is now spawned with
  `start_new_session=True`), escalating to SIGKILL on timeout.

Routed all three termination sites in `start_testgen_app` through the
helper (port-readiness timeout, KeyboardInterrupt handler, finally
cleanup). Added unit tests covering the platform branching and the
SIGKILL escalation path.

Note: this is a force-kill safety net for uncatchable termination paths
(installer crash, terminal X-button close, hang). Graceful shutdown on
Ctrl+C is a separate fix in pixeltable-pgserver / testgen's signal
handling — see the issue we drafted for the testgen team.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Author: aarthy-dk

dk-installer.py

@@ -20,6 +20,7 @@
 import re
 import secrets
 import shutil
+import signal
 import socket
 import ssl
 import stat
@@ -2464,6 +2465,40 @@ def read_testgen_config_env() -> dict[str, str]:
     return config
 
 
+def stop_app_tree(proc: subprocess.Popen, timeout: int = 10) -> None:
+    """Terminate ``proc`` and all of its descendants.
+
+    Plain ``proc.terminate()`` only kills the parent — pixeltable-pgserver
+    spawns ``postgres`` children that get orphaned otherwise. Cross-platform:
+    on Windows we shell out to ``taskkill /F /T``; on POSIX we send SIGTERM
+    to the whole process group (the parent was started with
+    ``start_new_session=True``).
+    """
+    if proc.poll() is not None:
+        return
+    if platform.system() == "Windows":
+        with contextlib.suppress(Exception):
+            subprocess.run(
+                ["taskkill", "/F", "/T", "/PID", str(proc.pid)],
+                stdout=subprocess.DEVNULL,
+                stderr=subprocess.DEVNULL,
+                creationflags=getattr(subprocess, "CREATE_NO_WINDOW", 0),
+                check=False,
+            )
+    else:
+        with contextlib.suppress(Exception):
+            os.killpg(os.getpgid(proc.pid), signal.SIGTERM)
+    try:
+        proc.wait(timeout=timeout)
+    except subprocess.TimeoutExpired:
+        if platform.system() != "Windows":
+            with contextlib.suppress(Exception):
+                os.killpg(os.getpgid(proc.pid), signal.SIGKILL)
+        proc.kill()
+        with contextlib.suppress(subprocess.TimeoutExpired):
+            proc.wait(timeout=5)
+
+
 def start_testgen_app(action, args) -> None:
     """Start ``testgen run-app`` and block until the user interrupts.
 
@@ -2496,15 +2531,17 @@ def start_testgen_app(action, args) -> None:
             stdout=subprocess.DEVNULL,
             stderr=subprocess.DEVNULL,
             stdin=subprocess.DEVNULL,
+            # POSIX-only: put the parent in its own process group so we can
+            # signal the whole tree (postgres included) on shutdown. Windows
+            # gets the same effect via ``taskkill /T`` in ``stop_app_tree``.
+            start_new_session=(platform.system() != "Windows"),
         )
     except FileNotFoundError as e:
         raise InstallerError(f"Could not start TestGen: {e}") from e
 
     try:
         if not wait_for_tcp_port(port, timeout=TESTGEN_APP_READY_TIMEOUT):
-            proc.terminate()
-            with contextlib.suppress(subprocess.TimeoutExpired):
-                proc.wait(timeout=5)
+            stop_app_tree(proc, timeout=5)
             raise InstallerError(
                 f"TestGen did not start within {TESTGEN_APP_READY_TIMEOUT} seconds. "
                 f"See {simplify_path(TESTGEN_LOG_FILE_PATH)} for details."
@@ -2526,19 +2563,11 @@ def start_testgen_app(action, args) -> None:
             # Reset the cursor to column 0 — the terminal echoed `^C` mid-line.
             print("")
             CONSOLE.msg("Stopping TestGen...")
-            proc.terminate()
-            try:
-                proc.wait(timeout=10)
-            except subprocess.TimeoutExpired:
-                proc.kill()
-                proc.wait()
+            stop_app_tree(proc, timeout=10)
             CONSOLE.msg("TestGen stopped.")
             CONSOLE.msg(f"To start it again, {command_hint(args.prod, 'start', 'Start TestGen')}.")
     finally:
-        if proc.poll() is None:
-            proc.terminate()
-            with contextlib.suppress(subprocess.TimeoutExpired):
-                proc.wait(timeout=5)
+        stop_app_tree(proc, timeout=5)
 
 
 class UvToolUpgradeStep(Step):

tests/test_tg_start.py

@@ -10,6 +10,7 @@
     InstallerError,
     TestgenStartAction,
     start_testgen_app,
+    stop_app_tree,
     InstallMarker,
 )
 
@@ -93,41 +94,114 @@ def test_start_testgen_app_aborts_on_port_timeout(app_action, args_mock, proc_ru
         patch("tests.installer.resolve_testgen_path", return_value="/bin/testgen"),
         patch("tests.installer.subprocess.Popen", return_value=proc_running_then_stops),
         patch("tests.installer.wait_for_tcp_port", return_value=False),
+        patch("tests.installer.stop_app_tree") as stop_mock,
         pytest.raises(InstallerError, match="did not start within"),
     ):
         start_testgen_app(app_action, args_mock)
 
-    proc_running_then_stops.terminate.assert_called()
+    stop_mock.assert_called_with(proc_running_then_stops, timeout=5)
 
 
 @pytest.mark.unit
 def test_start_testgen_app_handles_keyboard_interrupt(app_action, args_mock, console_msg_mock, empty_tg_config):
-    """User Ctrl+C during run is the expected stop signal — terminate cleanly,
-    don't propagate the exception, and hint at the start command for next time."""
+    """User Ctrl+C during run is the expected stop signal — kill the whole
+    process tree (postgres + parent), don't propagate the exception, and hint
+    at the start command for next time."""
     args_mock.prod = "tg"
 
     proc = MagicMock()
-    # poll() returns None while alive; after terminate() it transitions to 0.
     proc.poll.return_value = None
-
-    def _on_terminate():
-        proc.poll.return_value = 0
-
-    proc.terminate.side_effect = _on_terminate
     proc.wait.side_effect = [KeyboardInterrupt(), 0]
 
     with (
         patch("tests.installer.resolve_testgen_path", return_value="/bin/testgen"),
         patch("tests.installer.subprocess.Popen", return_value=proc),
         patch("tests.installer.wait_for_tcp_port", return_value=True),
+        patch("tests.installer.stop_app_tree") as stop_mock,
     ):
         start_testgen_app(app_action, args_mock)
 
-    proc.terminate.assert_called()
+    # Called once for the keyboard-interrupt branch (timeout=10) and again in
+    # the ``finally`` cleanup (timeout=5; no-op since proc already stopped).
+    assert stop_mock.call_args_list[0].args[0] is proc
+    assert stop_mock.call_args_list[0].kwargs == {"timeout": 10}
     console_msg_mock.assert_any_msg_contains("TestGen stopped")
     console_msg_mock.assert_any_msg_contains("tg start")
 
 
+# --- stop_app_tree ------------------------------------------------------------
+
+
+@pytest.mark.unit
+def test_stop_app_tree_no_op_when_proc_already_exited():
+    proc = MagicMock()
+    proc.poll.return_value = 0  # already exited
+
+    with patch("tests.installer.subprocess.run") as run_mock:
+        stop_app_tree(proc)
+
+    run_mock.assert_not_called()
+    proc.wait.assert_not_called()
+
+
+@pytest.mark.unit
+def test_stop_app_tree_windows_uses_taskkill_tree():
+    proc = MagicMock()
+    proc.poll.return_value = None
+    proc.pid = 4242
+    proc.wait.return_value = 0
+
+    with (
+        patch("tests.installer.platform.system", return_value="Windows"),
+        patch("tests.installer.subprocess.run") as run_mock,
+    ):
+        stop_app_tree(proc, timeout=3)
+
+    cmd = run_mock.call_args.args[0]
+    assert cmd[:4] == ["taskkill", "/F", "/T", "/PID"]
+    assert cmd[4] == "4242"
+    proc.wait.assert_called_with(timeout=3)
+
+
+@pytest.mark.unit
+def test_stop_app_tree_posix_signals_process_group():
+    proc = MagicMock()
+    proc.poll.return_value = None
+    proc.pid = 4242
+    proc.wait.return_value = 0
+
+    with (
+        patch("tests.installer.platform.system", return_value="Linux"),
+        patch("tests.installer.os.killpg") as killpg_mock,
+        patch("tests.installer.os.getpgid", return_value=4242),
+    ):
+        stop_app_tree(proc, timeout=3)
+
+    killpg_mock.assert_called_once()
+    assert killpg_mock.call_args.args[0] == 4242  # pgid
+    proc.wait.assert_called_with(timeout=3)
+
+
+@pytest.mark.unit
+def test_stop_app_tree_falls_through_to_kill_on_timeout():
+    """If SIGTERM doesn't take, escalate to SIGKILL / proc.kill()."""
+    import subprocess as sp
+
+    proc = MagicMock()
+    proc.poll.return_value = None
+    proc.pid = 4242
+    proc.wait.side_effect = [sp.TimeoutExpired(cmd="x", timeout=1), 0]
+
+    with (
+        patch("tests.installer.platform.system", return_value="Linux"),
+        patch("tests.installer.os.killpg"),
+        patch("tests.installer.os.getpgid", return_value=4242),
+    ):
+        stop_app_tree(proc, timeout=1)
+
+    proc.kill.assert_called_once()
+
+
 # --- TestgenStartAction -------------------------------------------------------