fix(install): make Ctrl+C abort indexing promptly

src/java_codebase_rag/cli.py

@@ -1162,8 +1162,20 @@ def _console_script_main() -> None:
     racy teardown — the command has already done its work and emitted its result.
     ``main()`` stays return-based so in-process test callers (``cli.main(...)``)
     keep working.
+
+    ``KeyboardInterrupt`` (Ctrl+C during a long indexing step) is caught here
+    rather than left to propagate: an uncaught interrupt bypasses this function
+    and runs full interpreter finalization (traceback + thread teardown),
+    whereas routing it through the same flush + ``os._exit`` path gives a clean,
+    immediate exit (code 130) and avoids the finalization-time SIGABRT noted
+    above for commands that loaded lancedb.
     """
-    rc = main()
+    try:
+        rc = main()
+    except KeyboardInterrupt:
+        sys.stderr.write("\nInterrupted.\n")
+        sys.stderr.flush()
+        rc = 130
     sys.stdout.flush()
     sys.stderr.flush()
     os._exit(rc)

src/java_codebase_rag/jrag.py

@@ -4551,9 +4551,19 @@ def _console_script_main() -> None:
     (SIGABRT, exit -6). Flushing + ``os._exit`` skips that racy teardown - the
     command has already done its work and emitted its result. ``main()`` stays
     return-based so in-process test callers keep working.
+
+    ``KeyboardInterrupt`` (Ctrl+C during a long indexing step) is caught here so
+    it routes through the same flush + ``os._exit`` path — clean, immediate exit
+    (code 130, no traceback) and no finalization-time SIGABRT — instead of
+    propagating past this function.
     """
     force_utf8_stdio()
-    rc = main()
+    try:
+        rc = main()
+    except KeyboardInterrupt:
+        sys.stderr.write("\nInterrupted.\n")
+        sys.stderr.flush()
+        rc = 130
     sys.stdout.flush()
     sys.stderr.flush()
     os._exit(rc)

src/java_codebase_rag/pipeline.py

@@ -133,14 +133,60 @@ def drain_err() -> None:
     t_err = threading.Thread(target=drain_err, name="stream-stderr", daemon=True)
     t_out.start()
     t_err.start()
+    # Wait on the CHILD before joining the drain threads. ``Popen.wait()`` is
+    # interruptible by Ctrl+C — the underlying ``os.waitpid`` returns EINTR and
+    # CPython raises ``KeyboardInterrupt`` — whereas ``Thread.join()`` blocks on
+    # an internal lock whose infinite-timeout acquire CPython never polls for
+    # signals. Joining *first* therefore made the whole indexing step ignore
+    # Ctrl+C until the child happened to close its pipes; cocoindex's teardown
+    # (and any flow-server grandchild it spawned) can hold them open for a long
+    # time, so an install/reprocess could not be aborted. The daemon drain
+    # threads keep the child's stdout/stderr pipes empty while we wait, so the
+    # child never blocks on a full pipe.
+    try:
+        code = proc.wait()
+    except BaseException:
+        # Ctrl+C or any other abort: best-effort, NON-BLOCKING child teardown
+        # so a process we spawned does not outlive us, then re-raise WITHOUT
+        # joining the drain threads. They may be blocked on a pipe the child
+        # still owns, and a join here would re-introduce the very hang this
+        # reordering fixes. The threads are daemons, so they vanish at exit.
+        _abort_child(proc)
+        raise
+    # Normal exit: the child is gone, its pipes hit EOF, and the drain threads
+    # return promptly — safe to join here. (If a pipe-inheriting grandchild
+    # lingered past the child's exit, these joins would block until it too
+    # closed its write ends — on Ctrl+C the shared-process-group SIGINT reaches
+    # it and closes them. This window is the short teardown-after-indexing phase,
+    # not the long indexing phase the wait-first reorder already made
+    # interruptible.)
     t_out.join()
     t_err.join()
     if filt is not None:
         filt.flush()
-    code = proc.wait()
     return out_buf.decode(errors="replace"), err_buf.decode(errors="replace"), code
 
 
+def _abort_child(proc: subprocess.Popen[bytes]) -> None:
+    """Best-effort, NON-BLOCKING teardown of a spawned child on an abort path.
+
+    Runs when ``proc.wait()`` raised (Ctrl+C, or any other exception). Sends
+    SIGTERM and returns immediately — this path exists to let the operator exit
+    *promptly*, and waiting for the child would defeat that. On Ctrl+C the child
+    already received SIGINT (same process group); this just guarantees teardown
+    for non-signal aborts, after which the child finishes shutting down on its
+    own. ``OSError`` (incl. ``ProcessLookupError`` — already-dead / zombie /
+    reaped) is swallowed — the only caller re-raises regardless.
+    """
+    fn = getattr(proc, "terminate", None)
+    if fn is None:
+        return
+    try:
+        fn()
+    except OSError:
+        pass
+
+
 def run_cocoindex_update(
     env: dict[str, str],
     *,

tests/package/test_java_codebase_rag_cli.py

@@ -1392,6 +1392,41 @@ def fake_exit(code: int) -> None:
         assert result is None, fake_rc
 
 
+def test_console_script_main_ctrl_c_exits_130_via_os_exit(
+    monkeypatch: pytest.MonkeyPatch,
+    capsys: pytest.CaptureFixture[str],
+) -> None:
+    """Ctrl+C (KeyboardInterrupt out of ``main``) must produce a clean, immediate
+    exit: code 130 (128 + SIGINT), a short notice on stderr (not stdout), and
+    routing through the same flush + ``os._exit`` path as the normal rc — never
+    an uncaught traceback into interpreter finalization. This is the operator-
+    facing contract for aborting a long indexing step, reachable once
+    ``pipeline._popen_capturing_stderr`` waits on the child before joining."""
+    import os as _os
+
+    from java_codebase_rag import cli as cli
+
+    snapshot: dict[str, object] = {}
+
+    def fake_main() -> int:
+        raise KeyboardInterrupt
+
+    monkeypatch.setattr(cli, "main", fake_main)
+
+    def fake_exit(code: int) -> None:
+        snapshot["exit_code"] = code
+
+    monkeypatch.setattr(_os, "_exit", fake_exit)
+
+    result = cli._console_script_main()
+
+    assert snapshot["exit_code"] == 130
+    assert result is None
+    captured = capsys.readouterr()
+    assert captured.out == "", "Ctrl+C must not write to stdout (machine consumers)"
+    assert "Interrupted" in captured.err, "Ctrl+C should print a short notice on stderr"
+
+
 def test_console_script_entry_point_routes_through_wrapper() -> None:
     """``[project.scripts]`` must point ``java-codebase-rag`` at
     ``_console_script_main`` (not ``main``) so the deterministic-exit path is the

tests/package/test_pipeline.py

@@ -5,10 +5,16 @@
 (cocoindex's bulk-update ``merge_insert``) emits ~one deletion-vector + version
 commit per matched row — O(rows) of tiny file IO that hangs for many minutes on
 large repos. Drop+recreate is identical output for a full rebuild.
+
+Also covers ``_popen_capturing_stderr``'s Ctrl+C behavior: it must wait on the
+child (interruptible) before joining the drain threads (not interruptible), and
+on abort must terminate the child and re-raise WITHOUT joining.
 """
 from __future__ import annotations
 
 import subprocess
+import sys
+import threading
 
 from java_codebase_rag import pipeline
 
@@ -101,3 +107,111 @@ def test_drop_preflight_blocker_is_silent(monkeypatch, capsys) -> None:
     assert seen["update"] == 1
     # 127 preflight is expected on graph-only installs and must NOT warn.
     assert "drop-before-reprocess failed" not in capsys.readouterr().err
+
+
+def test_popen_captures_normal_child_output() -> None:
+    """Regression guard: a completing child's stdout/stderr/code are still captured
+    after the wait-before-join reorder (verifies normal operation is unchanged)."""
+    proc = subprocess.Popen(
+        [
+            sys.executable,
+            "-c",
+            "import sys; sys.stdout.write('hello-out'); "
+            "sys.stderr.write('hello-err'); sys.exit(0)",
+        ],
+        stdout=subprocess.PIPE,
+        stderr=subprocess.PIPE,
+        bufsize=0,
+    )
+    try:
+        out, err, code = pipeline._popen_capturing_stderr(proc, verbose=True)
+    finally:
+        proc.wait()
+    assert code == 0
+    assert out == "hello-out"
+    assert "hello-err" in err
+
+
+def test_popen_abort_terminates_child_without_joining() -> None:
+    """Ctrl+C path: when ``proc.wait()`` raises while drain threads are blocked,
+    ``_popen_capturing_stderr`` must terminate the child and re-raise WITHOUT
+    joining the drain threads — a join here re-introduces the uninterruptible
+    hang (``Thread.join()`` on an infinite-timeout lock cannot be interrupted by
+    SIGINT), which is exactly the bug being fixed.
+
+    Run the helper in a daemon thread with a hard deadline so that a regression
+    (a join sneaking back onto the abort path) fails the assertion instead of
+    hanging the whole test session.
+    """
+    release = threading.Event()
+
+    class _BlockingStream:
+        """``proc.stdout``/``.stderr`` stand-in: ``read()`` blocks until released."""
+
+        def read(self, _n: int) -> bytes:
+            release.wait()
+            return b""
+
+    class _InterruptedProc:
+        def __init__(self) -> None:
+            self.stdout = _BlockingStream()
+            self.stderr = _BlockingStream()
+            self.terminated = False
+
+        def wait(self, *_a, **_k) -> int:
+            # Simulate Ctrl+C interrupting the (interruptible) wait on the child.
+            raise KeyboardInterrupt
+
+        def terminate(self) -> None:
+            self.terminated = True
+
+    proc = _InterruptedProc()
+    outcome: dict = {}
+
+    def runner() -> None:
+        try:
+            pipeline._popen_capturing_stderr(proc, verbose=True)
+            outcome["raised"] = None
+        except BaseException as exc:  # noqa: BLE001 — we WANT every escape
+            outcome["raised"] = exc
+        finally:
+            release.set()
+
+    t = threading.Thread(target=runner, daemon=True)
+    t.start()
+    t.join(timeout=5.0)
+    # Snapshot liveness BEFORE releasing: with the buggy join-first order the
+    # runner would still be blocked in Thread.join() here (deterministic); with
+    # the fix it returns in milliseconds. Checking after release.set() would race.
+    hung = t.is_alive()
+    release.set()  # never leave drain threads blocked, even on a passing run
+    t.join(timeout=5.0)
+
+    assert not hung, (
+        "_popen_capturing_stderr hung on the abort path — it must not join the "
+        "drain threads when the child wait is interrupted (Ctrl+C regression)"
+    )
+    assert isinstance(outcome.get("raised"), KeyboardInterrupt), (
+        "expected the KeyboardInterrupt to propagate out of the abort path"
+    )
+    assert proc.terminated, "the spawned child must be torn down on abort"
+
+
+def test_abort_child_swallows_already_dead() -> None:
+    """``_abort_child`` must not raise if the child is already gone (zombie /
+    reaped) — the only caller is on an exception path whose job is to re-raise,
+    so a teardown failure here must never replace the abort cause.
+
+    ``ProcessLookupError`` is the real-world shape (wait() raised, OS already
+    reaped the child between wait and terminate); it's a subclass of ``OSError``,
+    which is what the helper catches.
+    """
+    class _AlreadyDeadProc:
+        terminated = False
+
+        def terminate(self) -> None:
+            raise ProcessLookupError(3, "No such process")
+
+    proc = _AlreadyDeadProc()
+    # Must not raise — caller relies on best-effort, swallow-and-return.
+    pipeline._abort_child(proc)  # type: ignore[arg-type]