diff --git a/CHANGELOG.md b/CHANGELOG.md index 19c1bc1..bb1729d 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -10,6 +10,19 @@ Newest entries on top, within each tool. ## scriptkit +### 1.2.1 — 2026-07-03 +- **Fix: `parallel_map` now honors Ctrl-C promptly instead of hanging** (POK-85). + It ran on a `ThreadPoolExecutor` whose context-manager exit blocks until every + in-flight task finishes, so an interrupt stalled until the slowest worker + returned (e.g. a version check stuck in a 20s network timeout) — the interrupt + looked like a hang. Workers are now daemon threads and the main thread collects + results through a queue polled with a short timeout, so `KeyboardInterrupt` is + raised in the main thread right away and the interpreter can exit without + waiting on abandoned work. Public signature, return value (results in + completion order), progress rendering, and worker-exception propagation are + unchanged; new regression tests cover completion order, exception propagation, + and prompt-interrupt. + ### 1.2.0 — 2026-06-30 - **New `blocks` module** (`scriptkit.blocks`): `sk.ManagedBlock(begin, end)` plus the pure helpers `render_block` / `find_block` / `has_block` / `upsert_block` / @@ -62,6 +75,24 @@ Newest entries on top, within each tool. ## aikit +### 1.15.5 — 2026-07-03 +- **Fix: Ctrl-C during `aikit update` (and `install`) now exits cleanly and kills + the whole update tree** (POK-85). `run()` executed each update/install command + with `subprocess.run(shell=True, capture_output=True)`, which on interrupt only + killed the immediate shell — the real installer it spawned (npm/pip/curl and + their children) was orphaned and kept running, so aborting an update never truly + returned control (most visible under tmux). Captured commands now launch in + their own session/process group (`start_new_session`) with stdin closed, and an + interrupt or timeout tears down the **entire** group (SIGINT/SIGTERM → SIGKILL) + before re-raising, so `sk.run_cli` still exits `130`. aikit — not the terminal — + owns the teardown, so behavior is identical in tmux and a plain shell. A timeout + now also kills the whole tree instead of leaking a runaway process. Exit codes, + return shape, and output are unchanged; regression tests cover session + isolation, timeout tree-teardown, and interrupt tree-teardown. +- Depends on `scriptkit` 1.2.1, which fixes `parallel_map` so Ctrl-C during agent + discovery (`aikit`, `doctor`, `setup`, first run) no longer hangs until every + in-flight version check finishes. + ### 1.15.4 — 2026-07-02 - **Fix: five POK-65 agents falsely showed "needs auth" when authenticated** (POK-78). `auggie`, `devin`, `qodo`, `openinterpreter`, and `plandex` were registered without diff --git a/aikit b/aikit index 8ca4a2b..79f7a79 100755 --- a/aikit +++ b/aikit @@ -1,6 +1,6 @@ #!/usr/bin/env python3 """ -aikit — AI Coding Agent CLI Installer & Manager v1.15.4 +aikit — AI Coding Agent CLI Installer & Manager v1.15.5 Install, update, authenticate, and manage 31 AI coding agent CLIs from one tool. Architecture: @@ -52,7 +52,7 @@ from rich.progress import Progress, SpinnerColumn, TextColumn from rich.table import Table from rich.text import Text -__version__ = "1.15.4" +__version__ = "1.15.5" import scriptkit as sk @@ -1056,24 +1056,112 @@ def _npm_global_bin_dirs(): # Subprocess helpers # ============================================================================= +def _terminate_process_tree(proc, pgid, first_signal, *, grace=2.0): + """Tear down ``proc`` and every process it spawned, then reap it. + + ``run`` starts captured children with ``start_new_session=True`` so the + child leads its own process group (pgid == pid); signalling the *group* + therefore reaches the shell, the real installer it exec'd (npm/pip/curl/…), + and any grandchildren. We send ``first_signal`` (``SIGINT`` for a Ctrl-C, + ``SIGTERM`` for a timeout) so a well-behaved tree can unwind, then always + ``SIGKILL`` the whole group — a child that swallows the first signal, or a + backgrounded step the shell already ``wait``-ed past, must not outlive the + abort. Behaviour is identical in tmux and a plain terminal because aikit, + not the tty, owns the teardown. + + ``pgid`` is ``None`` for a non-isolated run (``capture=False`` or a platform + without process groups); we then tear down only the single process, never a + group, so we can't hit aikit itself. + """ + def _reap(timeout): + try: + proc.wait(timeout=timeout) + return True + except subprocess.TimeoutExpired: + return False + + if pgid is not None and hasattr(os, "killpg"): + def _signal_group(sig): + try: + os.killpg(pgid, sig) + except (ProcessLookupError, PermissionError, OSError): + pass + + _signal_group(first_signal) + _reap(grace) # let a well-behaved tree exit on its own + _signal_group(signal.SIGKILL) # then guarantee no member survives + _reap(grace) + return + + # Non-isolated / non-POSIX fallback: terminate just the direct child. + if proc.poll() is not None: + return + try: + proc.terminate() + except OSError: + return + if not _reap(grace): + try: + proc.kill() + except OSError: + pass + _reap(grace) + + def run(cmd, timeout=300, shell=True, capture=True, env=None): - """Run a shell command. Returns (exit_code, stdout, stderr).""" + """Run a shell command. Returns (exit_code, stdout, stderr). + + Captured commands run in their own session/process group + (``start_new_session``) with stdin closed, so an interrupt or timeout tears + down the *entire* subprocess tree rather than just the immediate shell. On + Ctrl-C we signal the whole group and re-raise, letting the single + termination point in ``sk.run_cli`` exit 130 — no installer subprocess + survives the interrupt, in tmux or a plain terminal alike. A timeout kills + the tree the same way and reports it instead of leaking a runaway process. + """ run_env = {**os.environ, **env} if env else None + pipe = subprocess.PIPE if capture else None + # Isolate captured commands in their own session so we can tear down the + # whole tree on abort — POSIX only (start_new_session/killpg are no-ops or + # absent elsewhere; there we fall back to single-process teardown). + isolate = capture and os.name == "posix" try: - result = subprocess.run( + proc = subprocess.Popen( cmd, shell=shell, - capture_output=capture, + stdin=subprocess.DEVNULL if capture else None, + stdout=pipe, + stderr=pipe, text=True, - timeout=timeout, env=run_env, + start_new_session=isolate, ) - return result.returncode, result.stdout.strip(), result.stderr.strip() + except Exception as e: + return -1, "", str(e) + + # With start_new_session the child leads its own group, so pgid == pid; + # capture it up front so teardown can reach the whole tree even after the + # group leader itself has exited. + pgid = proc.pid if isolate else None + + try: + stdout, stderr = proc.communicate(timeout=timeout) except subprocess.TimeoutExpired: - return -1, "", f"Command timed out after {timeout}s: {cmd}" + _terminate_process_tree(proc, pgid, signal.SIGTERM) + try: # drain whatever the child emitted before we killed it + stdout, stderr = proc.communicate(timeout=5) + except (subprocess.TimeoutExpired, ValueError): + stdout, stderr = "", "" + return -1, (stdout or "").strip(), f"Command timed out after {timeout}s: {cmd}" + except KeyboardInterrupt: + _terminate_process_tree(proc, pgid, signal.SIGINT) + raise except Exception as e: + _terminate_process_tree(proc, pgid, signal.SIGTERM) return -1, "", str(e) + return proc.returncode, (stdout or "").strip(), (stderr or "").strip() + def _npm_global_prefixes(): """npm global prefix directories aikit may install into or need to clean up.""" diff --git a/scriptkit/__init__.py b/scriptkit/__init__.py index 4c29e92..d7ad22d 100644 --- a/scriptkit/__init__.py +++ b/scriptkit/__init__.py @@ -69,7 +69,7 @@ truncate, ) -__version__ = "1.2.0" +__version__ = "1.2.1" __all__ = [ # submodules diff --git a/scriptkit/progress.py b/scriptkit/progress.py index 766f548..976dcf7 100644 --- a/scriptkit/progress.py +++ b/scriptkit/progress.py @@ -7,7 +7,8 @@ from __future__ import annotations -import concurrent.futures +import queue +import threading from contextlib import contextmanager from . import style @@ -78,6 +79,13 @@ def parallel_map(fn, items, description: str = "Working", max_workers: int = 8): """Run ``fn`` over ``items`` concurrently, showing combined progress. Returns results in completion order. Exceptions propagate from the worker. + + Workers are daemon threads and the main thread collects their results + through a queue polled with a short timeout, so a Ctrl-C is raised promptly + in the main thread and the interpreter can exit without waiting on in-flight + work. A plain ``ThreadPoolExecutor`` would instead block at shutdown until + every started task finished — e.g. a version check stuck in a 20s network + timeout — making the interrupt look like a hang (POK-85). """ items = list(items) results = [] @@ -109,15 +117,48 @@ def parallel_map(fn, items, description: str = "Working", max_workers: int = 8): progress_cm = None workers = max(1, min(max_workers, len(items))) - with concurrent.futures.ThreadPoolExecutor(max_workers=workers) as pool: - futures = [pool.submit(fn, it) for it in items] - if progress_cm is not None: - with progress_cm as progress: - task = progress.add_task(description, total=len(items)) - for fut in concurrent.futures.as_completed(futures): - results.append(fut.result()) - progress.advance(task) - else: - for fut in concurrent.futures.as_completed(futures): - results.append(fut.result()) + pending = queue.Queue() + for item in items: + pending.put(item) + completed = queue.Queue() # (ok: bool, payload: result | exception) + + def _worker(): + while True: + try: + item = pending.get_nowait() + except queue.Empty: + return + try: + completed.put((True, fn(item))) + except BaseException as exc: # surfaced in the main thread below + completed.put((False, exc)) + + for i in range(workers): + threading.Thread( + target=_worker, name=f"parallel_map-{i}", daemon=True + ).start() + + def _drain(on_done=None): + for _ in range(len(items)): + # Poll with a timeout so a pending SIGINT is raised here between + # waits instead of being held off by an untimed blocking get(). + while True: + try: + ok, payload = completed.get(timeout=0.1) + break + except queue.Empty: + continue + if not ok: + raise payload + results.append(payload) + if on_done is not None: + on_done() + + if progress_cm is not None: + with progress_cm as progress: + task = progress.add_task(description, total=len(items)) + _drain(lambda: progress.advance(task)) + else: + _drain() + return results diff --git a/tests/test_aikit_interrupt.py b/tests/test_aikit_interrupt.py new file mode 100644 index 0000000..bb2cc3f --- /dev/null +++ b/tests/test_aikit_interrupt.py @@ -0,0 +1,133 @@ +"""Interrupt / subprocess-lifecycle tests for aikit's ``run`` (POK-85). + +``aikit update`` couldn't be aborted cleanly with Ctrl-C: the update command is +run through ``run()``, and the previous implementation only killed the immediate +shell, orphaning the real installer tree it spawned. ``run()`` now isolates +captured children in their own session/process group and tears down the *whole* +group on interrupt or timeout, so nothing survives the abort — the same in tmux +or a plain terminal. + +These are POSIX tests (process groups / signals); skipped elsewhere. Liveness is +checked via a heartbeat file the grandchild keeps touching rather than +``os.kill(pid, 0)``, which can't tell a killed-but-unreaped zombie from a live +process once the parent shell is gone. +""" + +from __future__ import annotations + +import os +import signal +import subprocess +import sys +import time + +import pytest + +pytestmark = pytest.mark.skipif(os.name != "posix", reason="POSIX process groups") + + +@pytest.fixture +def aikit(tool_loader): + return tool_loader("aikit") + + +def _tree_cmd(started: str, heartbeat: str) -> str: + """Shell command that backgrounds a signal-ignoring grandchild. + + The grandchild ignores SIGINT/SIGTERM, records its pid in ``started``, then + touches ``heartbeat`` every 50ms for ~100s (a bounded comprehension keeps it + a single ``python -c`` statement). A frozen heartbeat therefore means the + grandchild was actually terminated — only ``SIGKILL`` to the whole group can + do that. ``& wait`` keeps the captured parent shell alive so the run reaches + its interrupt/timeout instead of returning on its own. + """ + body = ( + "import os,signal,time;" + "signal.signal(signal.SIGINT,signal.SIG_IGN);" + "signal.signal(signal.SIGTERM,signal.SIG_IGN);" + f"open({started!r},'w').write(str(os.getpid()));" + f"[(open({heartbeat!r},'w').write(str(time.time())),time.sleep(0.05)) " + "for _ in range(2000)]" + ) + return f"{sys.executable} -c {body!r} & wait" + + +def _wait_for(path: str, timeout: float = 5.0) -> None: + deadline = time.time() + timeout + while time.time() < deadline: + if os.path.exists(path): + return + time.sleep(0.02) + raise AssertionError(f"{path} was never created") + + +def _assert_heartbeat_stops(path: str, settle: float = 0.5, timeout: float = 5.0) -> None: + """Pass once ``path``'s mtime holds still for ``settle`` (writer is dead).""" + _wait_for(path) + deadline = time.time() + timeout + last = None + stable_since = None + while time.time() < deadline: + try: + mtime = os.stat(path).st_mtime_ns + except FileNotFoundError: + mtime = None + now = time.time() + if mtime == last: + stable_since = stable_since or now + if now - stable_since >= settle: + return + else: + last = mtime + stable_since = None + time.sleep(0.03) + raise AssertionError(f"heartbeat {path} kept advancing; process not terminated") + + +def test_run_captures_output(aikit): + assert aikit.run("echo hi") == (0, "hi", "") + assert aikit.run("echo out; echo err 1>&2; exit 3") == (3, "out", "err") + + +def test_run_captured_child_is_session_isolated(aikit): + """Captured runs get their own process group, so a group signal on teardown + can only reach the child tree, never aikit itself.""" + code, out, _ = aikit.run(f"{sys.executable} -c 'import os; print(os.getpgrp())'") + assert code == 0 + assert int(out) != os.getpgrp() + + +def test_run_timeout_terminates_whole_tree(aikit, tmp_path): + started = tmp_path / "started" + heartbeat = tmp_path / "hb" + t0 = time.perf_counter() + code, _, err = aikit.run(_tree_cmd(str(started), str(heartbeat)), timeout=0.5) + elapsed = time.perf_counter() - t0 + + assert code == -1 + assert "timed out" in err + assert elapsed < 4.0 + _assert_heartbeat_stops(str(heartbeat)) + + +def test_terminate_process_tree_kills_group_on_interrupt(aikit, tmp_path): + started = tmp_path / "started" + heartbeat = tmp_path / "hb" + proc = subprocess.Popen( + _tree_cmd(str(started), str(heartbeat)), + shell=True, + stdout=subprocess.DEVNULL, + stderr=subprocess.DEVNULL, + start_new_session=True, + ) + try: + _wait_for(str(started)) + # Exactly what run() does on Ctrl-C: signal the whole group and reap. + aikit._terminate_process_tree(proc, proc.pid, signal.SIGINT) + assert proc.poll() is not None # the shell itself was reaped + _assert_heartbeat_stops(str(heartbeat)) + finally: + try: + os.killpg(proc.pid, signal.SIGKILL) + except (ProcessLookupError, PermissionError, OSError): + pass diff --git a/tests/test_cli_progress_tables.py b/tests/test_cli_progress_tables.py index 037d46e..d7db8ca 100644 --- a/tests/test_cli_progress_tables.py +++ b/tests/test_cli_progress_tables.py @@ -101,6 +101,56 @@ def test_parallel_map_empty(no_color): assert progress.parallel_map(lambda x: x, []) == [] +def test_parallel_map_preserves_completion_order(no_color): + import time + + order = progress.parallel_map( + lambda d: (time.sleep(d / 100.0) or d), [5, 1, 3, 2], max_workers=4 + ) + assert order[0] == 1 and order[-1] == 5 + + +def test_parallel_map_worker_exception_propagates(no_color): + with pytest.raises(ZeroDivisionError): + progress.parallel_map(lambda x: 1 / 0 if x == 2 else x, [1, 2, 3], max_workers=3) + + +@pytest.mark.skipif(__import__("os").name != "posix", reason="POSIX signals") +def test_parallel_map_interrupt_is_prompt(no_color): + """Ctrl-C surfaces promptly instead of blocking on in-flight workers. + + Regression for POK-85: the old ThreadPoolExecutor implementation blocked at + shutdown until every started task finished (e.g. a 20s network timeout), so + the interrupt looked like a hang. Daemon workers + an interruptible + collection loop mean the interpreter can bail immediately. + """ + import os + import signal + import threading + import time + + slow_finished = [] + fired = threading.Event() + + def fn(x): + if x == 0: + fired.set() + os.kill(os.getpid(), signal.SIGINT) # signal handled on main thread + return x + time.sleep(5) # an in-flight worker we must NOT wait for + slow_finished.append(x) + return x + + start = time.perf_counter() + with pytest.raises(KeyboardInterrupt): + progress.parallel_map(fn, [0, 1, 2, 3], "interrupt", max_workers=4) + elapsed = time.perf_counter() - start + + assert fired.is_set() + assert elapsed < 3.0, f"interrupt took {elapsed:.2f}s; slow workers were awaited" + assert slow_finished == [], "in-flight workers should be abandoned, not awaited" + + def test_status_contextmanager(no_color, capsys): with progress.status("working") as st: st.update("still working")