fix(process-runtime): keep a short cleanup budget short

retire_owned_process floored the native-termination share at a 50 ms
grace, so a caller asking for 20 ms received more time than its own
deadline. Derive the inner budget from what is left of the caller's
deadline instead, and cover the sub-grace case so the shared entry point
keeps that promise.

Signed-off-by: ChengLong GU <88359731+gcl-coder@users.noreply.github.com>

Author: gcl-coder

loopx/extensions/process_runtime.py

@@ -402,13 +402,16 @@ def retire_owned_process(
     if not math.isfinite(timeout) or timeout <= 0:
         raise ValueError("native process cleanup timeout must be positive and finite")
     deadline = time.monotonic() + timeout
+    # Split one budget: native termination gets most of it, the remainder is
+    # left for pipe handles a Windows reader thread may still own. The inner
+    # budget is derived from what is left of the caller's deadline and never
+    # widened to a floor, so a sub-50 ms cleanup stays sub-50 ms.
+    terminate_timeout = min(timeout * 0.8, deadline - time.monotonic())
     try:
-        # Reserve part of the single cleanup budget for the pipe handles that a
-        # Windows reader thread may still own.
         force_terminate_process(
             process,
             process_tree_owned=process_tree_owned,
-            timeout=max(0.05, timeout * 0.8),
+            timeout=max(terminate_timeout, 1e-6),
         )
     finally:
         _close_owned_streams_before_deadline(process, deadline)

tests/extensions/test_process_runtime.py

@@ -17,6 +17,7 @@
 import loopx.extensions.process_runtime as process_runtime
 from loopx.extensions.process_runtime import (
     force_terminate_process,
+    retire_owned_process,
     remove_owned_tree,
     run_capped_process,
     terminate_process_tree,
@@ -224,6 +225,34 @@ def collect(*, timeout):
     observer.stdout.close.assert_called_once_with()
 
 
+class _NoStreamsProcess:
+    """Stand-in for a Popen whose pipes ``communicate`` already released."""
+
+    stdin = None
+    stdout = None
+    stderr = None
+
+
+def test_retire_owned_process_never_widens_a_sub_grace_budget(monkeypatch):
+    """A 20 ms cleanup stays inside 20 ms; it is not raised to a 50 ms floor."""
+
+    process = _NoStreamsProcess()
+    captured = {}
+
+    def record_termination(process, *, process_tree_owned, timeout):
+        captured["timeout"] = timeout
+        captured["owned"] = process_tree_owned
+
+    monkeypatch.setattr(process_runtime, "force_terminate_process", record_termination)
+    started = time.monotonic()
+    retire_owned_process(process, process_tree_owned=True, timeout=.02)
+    elapsed = time.monotonic() - started
+
+    assert captured["owned"] is True
+    assert 0 < captured["timeout"] <= .016
+    assert elapsed < .05
+
+
 @pytest.mark.parametrize("timeout", [0, -1, float("nan"), float("inf")])
 def test_force_cleanup_rejects_an_unbounded_or_invalid_budget(timeout):
     process = Mock(spec=subprocess.Popen)