fix(run[timeout]) Add timeout to legacy subprocess runner

why: Callers like vcspull sync need a wall-clock deadline on VCS subprocesses
so one unresponsive repo (credential prompt, stalled network, or a command
like `git remote show origin` that pipes thousands of lines through the
busy-poll loop) cannot freeze a whole batch. The legacy run() path had no
timeout support at all, making that class of hang impossible to recover from.

what:
- Add `timeout: float | None = None` kwarg to libvcs._internal.run.run.
- Introduce _wait_with_deadline which uses selectors to avoid blocking
  stderr reads and enforces the deadline without busy-polling.
- On timeout, SIGTERM the child, escalate to SIGKILL after a short grace,
  drain remaining output, and raise exc.CommandTimeoutError with the
  partial output so callers can diagnose what the process was doing.
- timeout=None preserves the pre-existing code path byte-for-byte so no
  existing caller changes behaviour.
- Tests cover: default/None behaviour, deadline firing within a bounded
  wall clock, partial stderr capture, and child reaping (no zombies).

Author: tony

src/libvcs/_internal/run.py

@@ -10,11 +10,14 @@
 
 from __future__ import annotations
 
+import contextlib
 import datetime
 import logging
 import os
+import selectors
 import subprocess
 import sys
+import time
 import typing as t
 from collections.abc import Iterable, Mapping, MutableMapping, Sequence
 
@@ -147,6 +150,7 @@ def run(
     log_in_real_time: bool = False,
     check_returncode: bool = True,
     callback: ProgressCallbackProtocol | None = None,
+    timeout: float | None = None,
 ) -> str:
     """Run a command.
 
@@ -186,6 +190,13 @@ def progress_cb(output, timestamp):
                 sys.stdout.flush()
             run(['git', 'pull'], callback=progress_cb)
 
+    timeout : float, optional
+        Seconds to wait before terminating the subprocess. When the deadline is
+        exceeded the process is sent ``SIGTERM`` (then ``SIGKILL`` after a short
+        grace period) and :class:`libvcs.exc.CommandTimeoutError` is raised with
+        any output collected so far. ``None`` (default) disables the deadline and
+        preserves the legacy behaviour of blocking until the process exits.
+
     Upcoming changes
     ----------------
     When minimum python >= 3.10, pipesize: int = -1 will be added after umask.
@@ -240,13 +251,21 @@ def progress_cb(output: t.AnyStr, timestamp: datetime.datetime) -> None:
     # connected to a pseudo-TTY, which would require significant changes
     # to how subprocess execution is handled.
 
-    while code is None:
-        code = proc.poll()
+    if timeout is None:
+        while code is None:
+            code = proc.poll()
 
-        if callback and callable(callback) and proc.stderr is not None:
-            line = console_to_str(proc.stderr.read(128))
-            if line:
-                callback(output=line, timestamp=datetime.datetime.now())
+            if callback and callable(callback) and proc.stderr is not None:
+                line = console_to_str(proc.stderr.read(128))
+                if line:
+                    callback(output=line, timestamp=datetime.datetime.now())
+    else:
+        code = _wait_with_deadline(
+            proc,
+            deadline=time.monotonic() + timeout,
+            callback=callback,
+            cmd=_stringify_command(normalized_args),
+        )
     if callback and callable(callback):
         callback(output="\r", timestamp=datetime.datetime.now())
 
@@ -272,3 +291,109 @@ def progress_cb(output: t.AnyStr, timestamp: datetime.datetime) -> None:
             cmd=_stringify_command(normalized_args),
         )
     return output
+
+
+#: Grace period after ``terminate()`` before escalating to ``kill()``.
+_TIMEOUT_KILL_GRACE_SECONDS = 0.5
+
+#: Upper bound on the ``selectors.select()`` wait inside the deadline loop.
+_TIMEOUT_POLL_INTERVAL_SECONDS = 0.1
+
+
+def _wait_with_deadline(
+    proc: subprocess.Popen[bytes],
+    *,
+    deadline: float,
+    callback: ProgressCallbackProtocol | None,
+    cmd: str | list[str],
+) -> int:
+    """Wait for ``proc`` to exit, enforcing a wall-clock deadline.
+
+    Uses :mod:`selectors` so the wait unblocks when stderr is readable, when the
+    child exits, or when the per-iteration poll interval expires -- whichever
+    comes first. When the deadline is exceeded, the subprocess is reaped and
+    :class:`libvcs.exc.CommandTimeoutError` is raised with whatever output was
+    collected before the timeout.
+    """
+    sel = selectors.DefaultSelector()
+    stderr_stream = proc.stderr
+    if stderr_stream is not None:
+        # Non-blocking reads so the selector loop never stalls on a short read
+        # from a stuck child (e.g. git waiting on network or credentials).
+        try:
+            os.set_blocking(stderr_stream.fileno(), False)
+        except (OSError, ValueError):
+            stderr_stream = None
+        else:
+            sel.register(stderr_stream, selectors.EVENT_READ)
+
+    partial_chunks: list[bytes] = []
+    try:
+        while True:
+            code = proc.poll()
+            if code is not None:
+                return code
+
+            remaining = deadline - time.monotonic()
+            if remaining <= 0:
+                _terminate_process(proc)
+                drained = _drain_stream(stderr_stream) + _drain_stream(proc.stdout)
+                message = console_to_str(b"".join(partial_chunks) + drained)
+                raise exc.CommandTimeoutError(
+                    output=message,
+                    returncode=proc.returncode,
+                    cmd=cmd,
+                )
+
+            wait = min(_TIMEOUT_POLL_INTERVAL_SECONDS, remaining)
+            events = sel.select(timeout=wait) if stderr_stream is not None else ()
+            if not events:
+                continue
+
+            for key, _mask in events:
+                if key.fileobj is not stderr_stream:
+                    continue
+                try:
+                    chunk = stderr_stream.read(128)
+                except (BlockingIOError, OSError):
+                    chunk = b""
+                if not chunk:
+                    continue
+                partial_chunks.append(chunk)
+                if callback is not None and callable(callback):
+                    callback(
+                        output=console_to_str(chunk),
+                        timestamp=datetime.datetime.now(),
+                    )
+    finally:
+        sel.close()
+
+
+def _terminate_process(proc: subprocess.Popen[bytes]) -> None:
+    """Terminate ``proc`` gracefully, falling back to ``kill`` on the grace."""
+    if proc.poll() is not None:
+        return
+    try:
+        proc.terminate()
+    except (OSError, ProcessLookupError):
+        return
+    try:
+        proc.wait(timeout=_TIMEOUT_KILL_GRACE_SECONDS)
+    except subprocess.TimeoutExpired:
+        with contextlib.suppress(OSError, ProcessLookupError):
+            proc.kill()
+        # If the child is still unreachable after SIGKILL, bail rather than
+        # block forever -- we've already signalled the user-facing timeout.
+        with contextlib.suppress(subprocess.TimeoutExpired):
+            proc.wait(timeout=_TIMEOUT_KILL_GRACE_SECONDS)
+
+
+def _drain_stream(stream: t.IO[bytes] | None) -> bytes:
+    """Best-effort read of any remaining bytes from a subprocess pipe."""
+    if stream is None:
+        return b""
+    try:
+        data = stream.read() or b""
+    except (BlockingIOError, OSError, ValueError):
+        data = b""
+    return data

tests/_internal/test_run.py

@@ -3,8 +3,15 @@
 from __future__ import annotations
 
 import pathlib
+import subprocess
+import sys
+import time
+import typing as t
 
-from libvcs._internal.run import _normalize_command_args
+import pytest
+
+from libvcs import exc
+from libvcs._internal.run import _normalize_command_args, run
 
 
 def test_normalize_command_args_keeps_scalar_string() -> None:
@@ -28,3 +35,91 @@ def test_normalize_command_args_coerces_pathlike() -> None:
 
     assert _normalize_command_args(path) == ["example"]
     assert _normalize_command_args([path, "status"]) == ["example", "status"]
+
+
+def test_run_without_timeout_matches_legacy_behavior() -> None:
+    """Leaving ``timeout=None`` preserves the pre-timeout call semantics."""
+    output = run([sys.executable, "-c", "print('hello'); print('world')"])
+
+    assert "hello" in output
+    assert "world" in output
+
+
+def test_run_raises_command_timeout_when_deadline_exceeded() -> None:
+    """A command sleeping past ``timeout`` raises ``CommandTimeoutError`` fast."""
+    started = time.monotonic()
+
+    with pytest.raises(exc.CommandTimeoutError) as excinfo:
+        run(
+            [sys.executable, "-c", "import time; time.sleep(10)"],
+            timeout=0.3,
+        )
+
+    elapsed = time.monotonic() - started
+
+    # Upper bound keeps the test honest: the deadline must actually fire, not
+    # fall through to the legacy unbounded wait.
+    assert elapsed < 2.5, f"timeout took too long: {elapsed:.2f}s"
+    assert isinstance(excinfo.value, exc.CommandError)
+
+
+def test_run_timeout_captures_partial_stderr_output() -> None:
+    """Output produced before the timeout is preserved on ``CommandTimeoutError``."""
+    script = (
+        "import sys, time;"
+        "sys.stderr.write('first\\n');"
+        "sys.stderr.flush();"
+        "time.sleep(10)"
+    )
+
+    with pytest.raises(exc.CommandTimeoutError) as excinfo:
+        run(
+            [sys.executable, "-c", script],
+            timeout=0.5,
+            log_in_real_time=True,
+        )
+
+    # The pre-timeout stderr chunk is forwarded to the callback and stashed in
+    # the exception's ``output`` so callers can diagnose what the process was
+    # doing before it was killed.
+    assert "first" in excinfo.value.output
+
+
+def test_run_timeout_reaps_child_process() -> None:
+    """Timed-out processes are terminated; no zombies remain in the group."""
+    script = "import time; time.sleep(10)"
+
+    captured: dict[str, t.Any] = {}
+    original_popen = subprocess.Popen
+
+    def _capturing_popen(*args: t.Any, **kwargs: t.Any) -> t.Any:
+        proc = original_popen(*args, **kwargs)
+        captured["proc"] = proc
+        return proc
+
+    # Attribute replacement keeps the test narrow -- we just need a handle on
+    # the Popen that ``run`` created so we can assert the child was actually
+    # reaped rather than abandoned (no zombie left behind).
+    subprocess.Popen = _capturing_popen  # type: ignore[misc,assignment]
+    try:
+        with pytest.raises(exc.CommandTimeoutError):
+            run([sys.executable, "-c", script], timeout=0.3)
+    finally:
+        subprocess.Popen = original_popen  # type: ignore[misc]
+
+    proc = captured["proc"]
+    # ``returncode`` is only populated once ``wait`` succeeds, so a populated
+    # value proves the timeout branch both killed and reaped the child.
+    assert proc.returncode is not None
+
+
+def test_run_without_timeout_completes_successfully() -> None:
+    """Regression guard: quick commands return output, no TimeoutError, no hang."""
+    output = run([sys.executable, "-c", "print('ok')"])
+    assert "ok" in output
+
+
+def test_run_timeout_none_is_the_default() -> None:
+    """Omitting ``timeout`` is equivalent to ``timeout=None``."""
+    output = run([sys.executable, "-c", "print('default')"], timeout=None)
+    assert "default" in output