fix(change-quality): hold every Git read of the scope to one budget

loopx/capabilities/change_quality/scope.py

@@ -1,14 +1,43 @@
 from __future__ import annotations
 
 import hashlib
+import os
 import subprocess
 from pathlib import Path
 from typing import Any
 
+from ...extensions.process_runtime import retire_owned_process
 
 CHANGE_QUALITY_SCOPE_SCHEMA_VERSION = "change_quality_scope_v0"
 
 
+# The scope reads a checkout the operator may not control. A repository blocked
+# on a stale index lock, a network mount or an oversized history must not hold
+# the receipt open with no answer, so every read shares one budget.
+GIT_PROBE_TIMEOUT_SECONDS = 10.0
+
+
+class GitProbeUnavailable(ValueError):
+    """Git did not answer inside the read budget, or could not start."""
+
+
+def _retire_owned_probe(process: subprocess.Popen[Any]) -> None:
+    """Stop the read's own session, then drop its pipes unread.
+
+    Git runs helpers -- fsmonitor hooks, pagers -- inside the session it starts
+    in, so killing the leader alone leaves them behind. An inherited pipe also
+    outlives the leader, so the pipes are closed instead of drained: a drain
+    would wait on work this call no longer owns.
+    """
+
+    try:
+        retire_owned_process(process, process_tree_owned=True)
+    except (OSError, RuntimeError, ValueError, subprocess.SubprocessError):
+        # Cleanup is best effort. The read's own timeout or cancellation is the
+        # result the caller must receive, even when termination is unconfirmed.
+        pass
+
+
 def _git(
     repo_root: Path,
     *args: str,
@@ -18,14 +47,35 @@ def _git(
     # text mode: a non-None encoding would silently flip bytes-mode callers
     # into text mode because CPython treats any of text/encoding/errors as
     # text-mode requests.
-    return subprocess.run(
-        ["git", "-C", str(repo_root), *args],
-        check=False,
-        text=text,
-        encoding="utf-8" if text else None,
-        errors="replace" if text else None,
-        stdout=subprocess.PIPE,
-        stderr=subprocess.PIPE,
+    argv = ["git", "-C", str(repo_root), *args]
+    budget = GIT_PROBE_TIMEOUT_SECONDS
+    try:
+        process = subprocess.Popen(
+            argv,
+            stdout=subprocess.PIPE,
+            stderr=subprocess.PIPE,
+            text=text,
+            encoding="utf-8" if text else None,
+            errors="replace" if text else None,
+            start_new_session=os.name == "posix",
+        )
+    except OSError as exc:
+        raise GitProbeUnavailable(f"git {' '.join(args)} could not start: {exc}") from exc
+    try:
+        stdout, stderr = process.communicate(timeout=budget)
+    except subprocess.TimeoutExpired:
+        _retire_owned_probe(process)
+        raise GitProbeUnavailable(
+            f"git {' '.join(args)} did not answer within {budget:.1f}s"
+        ) from None
+    except BaseException:
+        # A terminal Ctrl-C reaches this process's group, not the session the
+        # read started, so the cancellation retires its helpers here and then
+        # keeps travelling to its own handler.
+        _retire_owned_probe(process)
+        raise
+    return subprocess.CompletedProcess(
+        args=argv, returncode=process.returncode, stdout=stdout, stderr=stderr
     )
 
 

loopx/extensions/process_runtime.py

@@ -1,7 +1,5 @@
 from __future__ import annotations
 
-from collections.abc import Callable, Mapping, Sequence
-from dataclasses import dataclass
 import math
 import os
 import shutil
@@ -11,9 +9,10 @@
 import sys
 import threading
 import time
+from collections.abc import Callable, Mapping, Sequence
+from dataclasses import dataclass
 from pathlib import Path
-from typing import BinaryIO
-
+from typing import Any, BinaryIO
 
 _PROCESS_IO_CHUNK_BYTES = 64 * 1024
 _PROCESS_TERMINATE_GRACE_SECONDS = 1.0
@@ -310,7 +309,15 @@ def _force_terminate_windows_process_tree(
             stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL,
         )
         try:
-            if taskkill.wait(timeout=max(0, deadline - time.monotonic()) / 2) != 0:
+            try:
+                exit_code = taskkill.wait(
+                    timeout=max(0, deadline - time.monotonic()) / 2
+                )
+            except subprocess.TimeoutExpired as exc:
+                raise RuntimeError(
+                    "owned Windows process-tree cleanup unconfirmed"
+                ) from exc
+            if exit_code != 0:
                 raise RuntimeError("owned Windows process-tree cleanup unconfirmed")
         finally:
             # subprocess.run(timeout=...) internally waits without a timeout
@@ -345,6 +352,71 @@ def force_terminate_process(
 
 
 
+def _close_owned_streams_before_deadline(
+    process: subprocess.Popen[bytes] | subprocess.Popen[str], deadline: float,
+) -> None:
+    """Close a process's pipe ends without waiting on a blocked reader.
+
+    ``communicate`` retains reader threads on Windows; a descendant holding the
+    other end of a pipe can keep that reader, and the stream's lock, alive.
+    Closing from the caller would then wait past the read budget, so each close
+    runs on a daemon thread and is joined only until the deadline. A close that
+    cannot finish is left to the reader it is waiting for.
+    """
+
+    threads: list[threading.Thread] = []
+    for stream in (process.stdin, process.stdout, process.stderr):
+        if stream is None:
+            continue
+
+        def close(stream: Any = stream) -> None:
+            try:
+                stream.close()
+            except (OSError, ValueError):
+                pass
+
+        thread = threading.Thread(
+            target=close,
+            name="loopx-process-stream-close",
+            daemon=True,
+        )
+        thread.start()
+        threads.append(thread)
+    for thread in threads:
+        thread.join(timeout=max(0, deadline - time.monotonic()))
+
+
+def retire_owned_process(
+    process: subprocess.Popen[bytes] | subprocess.Popen[str], *,
+    process_tree_owned: bool, timeout: float = _PROCESS_TERMINATE_GRACE_SECONDS,
+) -> None:
+    """Force-retire an owned process and release its pipes within a deadline.
+
+    ``force_terminate_process`` bounds the native termination path, including
+    Windows taskkill and its helper. ``communicate`` may still retain a reader
+    thread, so pipe closure uses the same deadline and cannot turn a timed-out
+    read into an unbounded wait. The original timeout or cancellation remains
+    the caller's result even when termination cannot be confirmed.
+    """
+
+    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:
+        force_terminate_process(
+            process,
+            process_tree_owned=process_tree_owned,
+            timeout=max(terminate_timeout, 1e-6),
+        )
+    finally:
+        _close_owned_streams_before_deadline(process, deadline)
+
+
 def prepare_owned_process_cleanup(process: subprocess.Popen[bytes]) -> Callable[[], None]:
     """Retain diagnostic tree ownership even if its leader exits.
 

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)