diff --git a/CHANGELOG.md b/CHANGELOG.md
index ad06f2c..9fa0106 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -4,6 +4,66 @@ All notable changes to Code Context Control (C3) are documented here.
 The format is loosely based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/),
 and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
+## [2.67.1] - 2026-08-07
+
+### Fixed — a startup thread could wedge the whole MCP server, silently
+
+Three separate hangs, all of which left a process that looked healthy. The
+event loop stayed alive and idle throughout, so nothing logged, nothing
+crashed, and no health check noticed; the only visible symptom was every c3
+tool call dying at the client's 120s timeout.
+
+- **`collection.delete(where=...)` never returns.** `EmbeddingIndex.build()`
+  runs on the `c3-embed-index` thread spawned from `cli/mcp_server.py` and
+  calls `_remove_file_chunks()` for each changed file. That used chromadb's
+  `delete(where={"doc_id": ...})`, which was caught wedged inside the Rust
+  bindings (`chromadb/api/rust.py`, `RustBindingsAPI._delete`): two py-spy
+  dumps four minutes apart with byte-identical frames, 0.031s of CPU over
+  3s, and no writes to `chroma.sqlite3` for ten hours. `_remove_file_chunks`
+  now resolves the ids with `get(where=...)` first and deletes by explicit
+  id, moving the metadata filtering onto the read path, which does return.
+
+  The old code already carried a get-ids-then-delete fallback, and its
+  comment already suspected the where-delete — but the fallback sat behind
+  `except`, and an `except` clause cannot catch a call that never comes
+  back. It was unreachable by construction. It is now the only path.
+
+- **An unbounded lock turned one slow call into a dead server.** `build()`
+  took `self._lock` with a bare `with` and held it across the whole build
+  loop, so the wedged delete parked every later caller behind it forever. It
+  now acquires with a timeout (`_acquire_build_lock`, mirroring the
+  `_init_lock` pattern `_ensure_ready` already used correctly), logs once at
+  WARNING, and returns a `degraded` result carrying the normal stats shape so
+  callers reading it with `.get()` defaults keep working. A redundant build
+  is worth far less than a responsive server. This is the part that makes a
+  *future* backend hang survivable instead of fatal.
+
+- **`subprocess.run(timeout=...)` hangs inside its own timeout handler.**
+  `check_gemini` / `check_codex` / `check_claude` passed `stdin=DEVNULL` and
+  `timeout=10` and hung anyway: on Windows, when the timeout fires, CPython's
+  handler kills only the direct child and then calls `communicate()` a
+  *second* time with **no timeout** (the `_mswindows` branch of `run()` in
+  `Lib/subprocess.py`). That join never completes while a surviving
+  grandchild still holds the stdout/stderr write-ends. Observed wedging the
+  `c3-delegate-prewarm` thread for 10h and leaking its two reader threads, so
+  delegate health checks never completed and every first `c3_agent` call paid
+  full preflight. All three now go through `_probe_cli_version`: Popen, a
+  `taskkill /T` process-*tree* kill, and a bounded `communicate()` in
+  `finally`. `tests/test_cli_smoke.py` documented this exact footgun for test
+  code back in 2.43.0; production code now follows the same convention.
+
+### Tests
+
+- `tests/test_embedding_index_deadlock.py` — pins the chunk-removal contract
+  (a `doc_id`'s chunks are removed without ever passing `where=` to
+  `delete`), reproduces the hang against a blocking fake collection with a
+  bounded join so a regression *fails* rather than wedging pytest, and proves
+  the busy-lock path degrades instead of blocking.
+- `tests/test_delegate_version_probe.py` — asserts the second `communicate()`
+  is bounded, that the kill is a tree kill, and that the pipes close on every
+  exit path.
+- 22 of the 26 new tests fail against the pre-fix source.
+
 ## [2.67.0] - 2026-07-31
 
 ### Added — The Discipline tab grows search, evidence, and controls (Hub)
diff --git a/cli/c3.py b/cli/c3.py
index aacf063..0d898d8 100644
--- a/cli/c3.py
+++ b/cli/c3.py
@@ -92,7 +92,7 @@
 # Config
 CONFIG_DIR = ".c3"
 CONFIG_FILE = ".c3/config.json"
-__version__ = "2.67.0"
+__version__ = "2.67.1"
 
 
 def _compress_file_cli(compressor, path, mode="smart", **kw):
diff --git a/cli/tools/delegate.py b/cli/tools/delegate.py
index f79afef..d58f8ca 100644
--- a/cli/tools/delegate.py
+++ b/cli/tools/delegate.py
@@ -112,6 +112,60 @@ def _popen_kwargs():
     return kwargs
 
 
+def _probe_cli_version(exe: str, timeout: int = 10):
+    """Run ``<exe> --version`` without subprocess.run's timeout footgun.
+
+    ``subprocess.run(cmd, capture_output=True, timeout=N)`` reads as safe and
+    is not. When the timeout fires on Windows, CPython's own handler kills the
+    direct child and then calls ``process.communicate()`` a *second* time with
+    **no timeout** (the ``_mswindows`` branch of ``run()`` in Lib/subprocess.py).
+    That second call joins the stdout/stderr reader threads, which never see
+    EOF while any surviving grandchild still holds the pipe write-ends — so
+    ``run()`` blocks forever inside its own timeout handler. Observed wedging
+    the c3-delegate-prewarm thread for 10h and leaking its two reader threads,
+    so delegate health checks never completed and every first c3_agent call
+    paid full preflight.
+
+    Popen + a process-*tree* kill + a bounded communicate() in ``finally``
+    closes the write-ends no matter which way we leave. ``_kill_proc_tree``
+    uses ``taskkill /T`` on Windows, so grandchildren die too — the exact case
+    CPython's bare ``process.kill()`` misses.
+
+    Returns (stdout, stderr, returncode), or None if it timed out.
+    """
+    timed_out = False
+    proc = subprocess.Popen(
+        harden_win_argv([exe, "--version"]),
+        stdout=subprocess.PIPE, stderr=subprocess.PIPE,
+        stdin=subprocess.DEVNULL,
+        text=True, encoding="utf-8", errors="replace",
+        **_popen_kwargs(),
+    )
+    try:
+        try:
+            out, err = proc.communicate(timeout=timeout)
+        except subprocess.TimeoutExpired:
+            timed_out = True
+            return None
+        return (out or "").strip(), (err or "").strip(), proc.returncode
+    finally:
+        if proc.poll() is None:
+            _kill_proc_tree(proc)
+        if timed_out:
+            # Reap the reader threads with a bound. This is the call CPython
+            # makes with no timeout at all; the bound is the whole fix.
+            try:
+                proc.communicate(timeout=5)
+            except Exception:
+                pass
+            for stream in (proc.stdout, proc.stderr):
+                try:
+                    if stream is not None:
+                        stream.close()
+                except Exception:
+                    pass
+
+
 # ---------------------------------------------------------------------------
 # Codex CLI backend
 # ---------------------------------------------------------------------------
@@ -198,19 +252,16 @@ def check_claude() -> dict:
         _claude_available = False
         return {"status": "not_installed", "detail": "claude CLI not found on PATH"}
     try:
-        proc = subprocess.run(
-            [exe, "--version"],
-            capture_output=True, text=True, timeout=10,
-            stdin=subprocess.DEVNULL,
-        )
-        if proc.returncode == 0:
+        probed = _probe_cli_version(exe, timeout=10)
+        if probed is None:
+            _claude_available = False
+            return {"status": "timeout", "detail": "claude --version timed out (10s)"}
+        out, err, code = probed
+        if code == 0:
             _claude_available = True
-            return {"status": "ok", "version": proc.stdout.strip()}
-        _claude_available = False
-        return {"status": "error", "detail": proc.stderr.strip() or f"exit {proc.returncode}"}
-    except subprocess.TimeoutExpired:
+            return {"status": "ok", "version": out}
         _claude_available = False
-        return {"status": "timeout", "detail": "claude --version timed out (10s)"}
+        return {"status": "error", "detail": err or f"exit {code}"}
     except Exception as e:
         _claude_available = False
         return {"status": "error", "detail": str(e)}
@@ -286,21 +337,17 @@ def check_gemini() -> dict:
         _gemini_available = False
         return {"status": "not_installed", "detail": "gemini CLI not found on PATH"}
     try:
-        proc = subprocess.run(
-            [exe, "--version"],
-            capture_output=True, text=True, timeout=10,
-            stdin=subprocess.DEVNULL,
-        )
-        if proc.returncode == 0:
-            version = proc.stdout.strip()
+        probed = _probe_cli_version(exe, timeout=10)
+        if probed is None:
+            _gemini_available = False
+            return {"status": "timeout", "detail": "gemini --version timed out (10s)"}
+        out, err, code = probed
+        if code == 0:
             _gemini_available = True
-            return {"status": "ok", "version": version}
+            return {"status": "ok", "version": out}
         else:
             _gemini_available = False
-            return {"status": "error", "detail": proc.stderr.strip() or f"exit code {proc.returncode}"}
-    except subprocess.TimeoutExpired:
-        _gemini_available = False
-        return {"status": "timeout", "detail": "gemini --version timed out (10s)"}
+            return {"status": "error", "detail": err or f"exit code {code}"}
     except Exception as e:
         _gemini_available = False
         return {"status": "error", "detail": str(e)}
@@ -545,21 +592,17 @@ def check_codex() -> dict:
         _codex_available = False
         return {"status": "not_installed", "detail": "codex CLI not found on PATH"}
     try:
-        proc = subprocess.run(
-            [exe, "--version"],
-            capture_output=True, text=True, timeout=10,
-            stdin=subprocess.DEVNULL,
-        )
-        if proc.returncode == 0:
-            version = proc.stdout.strip()
+        probed = _probe_cli_version(exe, timeout=10)
+        if probed is None:
+            _codex_available = False
+            return {"status": "timeout", "detail": "codex --version timed out (10s)"}
+        out, err, code = probed
+        if code == 0:
             _codex_available = True
-            return {"status": "ok", "version": version}
+            return {"status": "ok", "version": out}
         else:
             _codex_available = False
-            return {"status": "error", "detail": proc.stderr.strip() or f"exit code {proc.returncode}"}
-    except subprocess.TimeoutExpired:
-        _codex_available = False
-        return {"status": "timeout", "detail": "codex --version timed out (10s)"}
+            return {"status": "error", "detail": err or f"exit code {code}"}
     except Exception as e:
         _codex_available = False
         return {"status": "error", "detail": str(e)}
diff --git a/pyproject.toml b/pyproject.toml
index b7565b6..378f27e 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "code-context-control"
-version = "2.67.0"
+version = "2.67.1"
 description = "Local MCP code-intelligence for AI coding tools: surgical search/read/edit, agent-config version history, path-level access + masking guards, and a multi-project hub."
 readme = "README.md"
 requires-python = ">=3.10"
diff --git a/services/embedding_index.py b/services/embedding_index.py
index f4fb552..695a589 100644
--- a/services/embedding_index.py
+++ b/services/embedding_index.py
@@ -14,6 +14,10 @@
 
 log = logging.getLogger("c3.embedding_index")
 _SEARCH_INIT_WAIT_SECONDS = 0.25
+# Upper bound on how long a caller will park waiting for an in-flight build.
+# A redundant build is worth far less than a responsive server, so we give up
+# and degrade rather than block. See _acquire_build_lock().
+_BUILD_LOCK_WAIT_SECONDS = 30.0
 
 
 class EmbeddingIndex:
@@ -43,6 +47,7 @@ def __init__(
         self._model_ok = False
         self._file_hashes: dict[str, str] = {}  # doc_id -> content hash
         self._lock = threading.Lock()
+        self._lock_warned = False  # WARN once, not once per blocked caller
         self._chunk_map: dict[str, dict] = {}  # chunk_id -> metadata
 
         # Heavy backend init (chromadb import/client + ollama probe) and hash
@@ -169,6 +174,45 @@ def _content_hash(content: str) -> str:
 
     # ── Build / Update ────────────────────────────────────
 
+    def _acquire_build_lock(self, timeout: float | None = None) -> bool:
+        """Acquire the build lock with a bound. Mirrors ``_ensure_ready``.
+
+        An unbounded ``with self._lock`` turns any slow backend call into a
+        dead server: whoever holds the lock never returns, every later caller
+        parks behind it forever, and the MCP client kills the tool call at its
+        own timeout while the event loop still looks perfectly healthy. A
+        bounded acquire degrades instead — the caller skips the embedding work
+        and serves its request without it.
+
+        Returns True when the lock is held (caller MUST release it).
+        """
+        wait = _BUILD_LOCK_WAIT_SECONDS if timeout is None else timeout
+        if self._lock.acquire(timeout=max(0.0, wait)):
+            return True
+        if not self._lock_warned:
+            self._lock_warned = True
+            log.warning(
+                "Embedding index build lock still held after %.1fs — skipping "
+                "this build. Semantic search keeps serving whatever is already "
+                "indexed; this is logged once per index instance.",
+                wait,
+            )
+        return False
+
+    def _busy_result(self) -> dict:
+        """Build stats shaped like a normal return, marked degraded."""
+        return {
+            "error": "Embedding index busy (build already in flight); skipped",
+            "available": True,
+            "degraded": True,
+            "files_processed": 0,
+            "files_skipped": 0,
+            "chunks_embedded": 0,
+            "chunks_skipped": 0,
+            "errors": 0,
+            "total_embedded": 0,
+        }
+
     def build(self, code_index, force: bool = False, on_progress=None) -> dict:
         """Build or incrementally update the embedding index from CodeIndex chunks.
 
@@ -213,7 +257,9 @@ def _report():
                 except Exception:
                     pass
 
-        with self._lock:
+        if not self._acquire_build_lock():
+            return self._busy_result()
+        try:
             # Detect deleted files — remove their embeddings
             indexed_files = set(self._file_hashes.keys())
             current_files = set(chunks_by_file.keys())
@@ -283,6 +329,8 @@ def _report():
                 _report()
 
             self._save_hashes()
+        finally:
+            self._lock.release()
 
         return {
             "files_processed": files_processed,
@@ -311,20 +359,41 @@ def _embed_batch(self, ids: list, texts: list, metas: list) -> bool:
             return False
 
     def _remove_file_chunks(self, doc_id: str):
-        """Remove all embedded chunks belonging to a file."""
+        """Remove all embedded chunks belonging to a file.
+
+        Resolves the ids first and deletes by id. It never calls
+        ``delete(where=...)``.
+
+        ``collection.delete(where=...)`` has been observed to never return
+        inside the chromadb Rust bindings (``chromadb/api/rust.py``,
+        ``RustBindingsAPI._delete``): two py-spy dumps four minutes apart with
+        byte-identical frames, 0.031s of CPU over 3s, and no writes to
+        chroma.sqlite3 for ten hours. Because this is a *hang* and not an
+        exception, the ``except``-guarded fallback that used to live here could
+        never fire — the comment right below it already suspected the
+        where-delete, but an ``except`` clause cannot catch a thread that never
+        comes back. Worse, build() called this while holding the build lock, so
+        one wedged delete took every later caller down with it.
+
+        Deleting by explicit id keeps the metadata filtering on ``get()``,
+        which does return, and leaves ``delete()`` with the one argument shape
+        that has never been seen to stall.
+        """
         if not self._collection:
             return
         try:
-            self._collection.delete(where={"doc_id": doc_id})
-        except Exception:
-            # Some chromadb versions don't support where-delete well;
-            # fall back to getting IDs first
             try:
-                results = self._collection.get(where={"doc_id": doc_id})
-                if results and results.get("ids"):
-                    self._collection.delete(ids=results["ids"])
+                # include=[] skips fetching documents/embeddings we throw away.
+                results = self._collection.get(
+                    where={"doc_id": doc_id}, include=[])
             except Exception:
-                pass
+                # Older chromadb (we support >=0.4.24) may reject include=[].
+                results = self._collection.get(where={"doc_id": doc_id})
+            ids = (results or {}).get("ids") or []
+            if ids:
+                self._collection.delete(ids=ids)
+        except Exception as e:
+            log.debug("Removing chunks for %s failed: %s", doc_id, e)
 
     # ── Search ────────────────────────────────────────────
 
diff --git a/tests/test_delegate_version_probe.py b/tests/test_delegate_version_probe.py
new file mode 100644
index 0000000..ff8f6c7
--- /dev/null
+++ b/tests/test_delegate_version_probe.py
@@ -0,0 +1,247 @@
+"""Regression tests for the CLI health-check subprocess hang.
+
+check_gemini/check_codex/check_claude used subprocess.run(capture_output=True,
+timeout=10). That looks safe and is not: when the timeout fires on Windows,
+CPython's own handler kills only the DIRECT child and then calls
+process.communicate() a SECOND time with no timeout (the _mswindows branch of
+run() in Lib/subprocess.py). That join never finishes while a surviving
+grandchild still holds the stdout/stderr write-ends, so run() hangs inside its
+own timeout handler -- observed wedging the c3-delegate-prewarm thread for 10h
+and leaking its two reader threads.
+
+tests/test_cli_smoke.py already documents this footgun and works around it for
+test code; _probe_cli_version applies the same convention to production code.
+"""
+import subprocess
+import sys
+
+import pytest
+
+from cli.tools import delegate
+
+# ---------------------------------------------------------------------------
+# Fakes
+# ---------------------------------------------------------------------------
+
+
+class _FakeStream:
+    def __init__(self):
+        self.closed = False
+
+    def close(self):
+        self.closed = True
+
+
+class _TimingOutPopen:
+    """Popen whose first communicate() times out, like a wedged CLI."""
+
+    instances = []
+
+    def __init__(self, *args, **kwargs):
+        self.args = args
+        self.kwargs = kwargs
+        self.pid = 424242
+        self.returncode = None
+        self.stdout = _FakeStream()
+        self.stderr = _FakeStream()
+        self.communicate_timeouts = []
+        self.alive = True
+        _TimingOutPopen.instances.append(self)
+
+    def communicate(self, input=None, timeout=None):
+        self.communicate_timeouts.append(timeout)
+        if len(self.communicate_timeouts) == 1:
+            raise subprocess.TimeoutExpired(cmd="fake --version", timeout=timeout)
+        if timeout is None:
+            # This is precisely the CPython behaviour we are avoiding.
+            raise AssertionError(
+                "second communicate() must be bounded, not open-ended"
+            )
+        return "", ""
+
+    def poll(self):
+        return None if self.alive else 0
+
+    def kill(self):
+        self.alive = False
+        self.returncode = -9
+
+    def wait(self, timeout=None):
+        return self.returncode
+
+
+@pytest.fixture
+def timing_out_popen(monkeypatch):
+    _TimingOutPopen.instances = []
+    killed = []
+
+    def _fake_kill_tree(proc):
+        killed.append(proc)
+        proc.alive = False
+        proc.returncode = -9
+
+    monkeypatch.setattr(delegate.subprocess, "Popen", _TimingOutPopen)
+    monkeypatch.setattr(delegate, "_kill_proc_tree", _fake_kill_tree)
+    return killed
+
+
+# ---------------------------------------------------------------------------
+# _probe_cli_version
+# ---------------------------------------------------------------------------
+
+
+def test_probe_returns_version_for_a_responsive_binary():
+    """Happy path against a real process: `python --version`."""
+    probed = delegate._probe_cli_version(sys.executable, timeout=30)
+
+    assert probed is not None, "probe timed out on a trivially fast command"
+    out, err, code = probed
+    assert code == 0
+    # Older CPython printed the version on stderr; accept either stream.
+    assert "Python" in (out or "") + (err or "")
+
+
+def test_probe_passes_devnull_stdin_and_pipes():
+    """stdin must never be inherited -- an interactive CLI would block."""
+    captured = {}
+
+    class _Recorder(_TimingOutPopen):
+        def __init__(self, *args, **kwargs):
+            captured.update(kwargs)
+            captured["argv"] = args[0] if args else None
+            super().__init__(*args, **kwargs)
+
+        def communicate(self, input=None, timeout=None):
+            self.communicate_timeouts.append(timeout)
+            self.alive = False
+            self.returncode = 0
+            return "v1.2.3\n", ""
+
+    real_popen = delegate.subprocess.Popen
+    delegate.subprocess.Popen = _Recorder
+    try:
+        probed = delegate._probe_cli_version("somecli", timeout=10)
+    finally:
+        delegate.subprocess.Popen = real_popen
+
+    assert probed == ("v1.2.3", "", 0)
+    assert captured["stdin"] is subprocess.DEVNULL
+    assert captured["stdout"] is subprocess.PIPE
+    assert captured["stderr"] is subprocess.PIPE
+
+
+def test_probe_timeout_kills_the_tree_and_bounds_the_second_communicate(
+    timing_out_popen,
+):
+    """The whole point: no open-ended communicate(), and a TREE kill."""
+    killed = timing_out_popen
+
+    probed = delegate._probe_cli_version("hangingcli", timeout=1)
+
+    assert probed is None, "a timed-out probe must report timeout, not a version"
+
+    proc = _TimingOutPopen.instances[-1]
+    # Two communicate() calls: the bounded first, and a bounded reap.
+    assert len(proc.communicate_timeouts) == 2
+    assert proc.communicate_timeouts[0] == 1
+    assert proc.communicate_timeouts[1] is not None, (
+        "second communicate() was open-ended -- this is the 10h hang"
+    )
+    # Tree kill, not Popen.kill(): grandchildren hold the pipe write-ends.
+    assert killed == [proc], "expected _kill_proc_tree, not a bare kill()"
+    # Pipes closed regardless of which way we left.
+    assert proc.stdout.closed and proc.stderr.closed
+
+
+def test_probe_does_not_leak_reader_threads_on_timeout(timing_out_popen):
+    """A wedged probe must not accumulate handles across repeated calls."""
+    for _ in range(5):
+        assert delegate._probe_cli_version("hangingcli", timeout=1) is None
+
+    assert len(_TimingOutPopen.instances) == 5
+    for proc in _TimingOutPopen.instances:
+        assert proc.stdout.closed and proc.stderr.closed
+        assert not proc.alive
+
+
+# ---------------------------------------------------------------------------
+# The health checks that the prewarm thread calls
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.parametrize(
+    "check_name,cli_name",
+    [("check_gemini", "gemini"), ("check_codex", "codex"), ("check_claude", "claude")],
+)
+def test_health_checks_report_timeout_without_hanging(
+    check_name, cli_name, timing_out_popen, monkeypatch
+):
+    monkeypatch.setattr(delegate, "_which", lambda name: f"/usr/bin/{name}")
+
+    result = getattr(delegate, check_name)()
+
+    assert result["status"] == "timeout"
+    assert cli_name in result["detail"]
+
+
+@pytest.mark.parametrize(
+    "check_name,flag",
+    [
+        ("check_gemini", "_gemini_available"),
+        ("check_codex", "_codex_available"),
+        ("check_claude", "_claude_available"),
+    ],
+)
+def test_health_checks_report_ok_and_set_availability(
+    check_name, flag, monkeypatch
+):
+    monkeypatch.setattr(delegate, "_which", lambda name: f"/usr/bin/{name}")
+    monkeypatch.setattr(
+        delegate, "_probe_cli_version", lambda exe, timeout=10: ("1.2.3", "", 0)
+    )
+
+    result = getattr(delegate, check_name)()
+
+    assert result == {"status": "ok", "version": "1.2.3"}
+    assert getattr(delegate, flag) is True
+
+
+@pytest.mark.parametrize(
+    "check_name", ["check_gemini", "check_codex", "check_claude"]
+)
+def test_health_checks_report_nonzero_exit_as_error(check_name, monkeypatch):
+    monkeypatch.setattr(delegate, "_which", lambda name: f"/usr/bin/{name}")
+    monkeypatch.setattr(
+        delegate, "_probe_cli_version", lambda exe, timeout=10: ("", "boom", 3)
+    )
+
+    result = getattr(delegate, check_name)()
+
+    assert result["status"] == "error"
+    assert result["detail"] == "boom"
+
+
+@pytest.mark.parametrize(
+    "check_name", ["check_gemini", "check_codex", "check_claude"]
+)
+def test_health_checks_report_not_installed(check_name, monkeypatch):
+    monkeypatch.setattr(delegate, "_which", lambda name: None)
+
+    result = getattr(delegate, check_name)()
+
+    assert result["status"] == "not_installed"
+
+
+def test_no_health_check_uses_subprocess_run():
+    """subprocess.run(timeout=) is the footgun; keep it out of these paths."""
+    import inspect
+
+    for name in ("check_gemini", "check_codex", "check_claude"):
+        src = inspect.getsource(getattr(delegate, name))
+        assert "subprocess.run" not in src, (
+            f"{name} went back to subprocess.run -- see _probe_cli_version"
+        )
+
+
+if __name__ == "__main__":
+    raise SystemExit(pytest.main([__file__, "-v"]))
diff --git a/tests/test_embedding_index_deadlock.py b/tests/test_embedding_index_deadlock.py
new file mode 100644
index 0000000..656f2e4
--- /dev/null
+++ b/tests/test_embedding_index_deadlock.py
@@ -0,0 +1,321 @@
+"""Regression tests for the startup deadlock in services/embedding_index.py.
+
+Two independent failures used to combine into a dead-but-healthy-looking MCP
+server:
+
+1. ``_remove_file_chunks`` called ``collection.delete(where=...)``, which was
+   observed never returning inside the chromadb Rust bindings
+   (``chromadb/api/rust.py``, ``RustBindingsAPI._delete``). The old
+   ``except``-guarded fallback could not help: a hang raises nothing, so the
+   ``except`` clause was unreachable by construction.
+
+2. ``build()`` held ``self._lock`` unconditionally while doing that, so the
+   wedged thread parked every later caller behind it forever.
+
+These tests pin both contracts. The hang is simulated with a blocking fake
+collection so a regression FAILS (in a daemon thread, bounded join) instead of
+hanging the suite -- the same convention tests/test_cli_smoke.py uses.
+"""
+import logging
+import threading
+
+import pytest
+
+from services import embedding_index as ei_mod
+from services.embedding_index import EmbeddingIndex
+
+# ---------------------------------------------------------------------------
+# Fakes
+# ---------------------------------------------------------------------------
+
+
+class _RecordingCollection:
+    """Records every delete/get call so tests can assert on their shape."""
+
+    def __init__(self, docs=None):
+        # id -> doc_id
+        self.docs = dict(docs or {})
+        self.delete_calls = []
+        self.get_calls = []
+
+    def get(self, where=None, include=None, **kwargs):
+        self.get_calls.append({"where": where, "include": include})
+        doc_id = (where or {}).get("doc_id")
+        ids = [i for i, d in self.docs.items() if d == doc_id]
+        return {"ids": ids}
+
+    def delete(self, ids=None, where=None, **kwargs):
+        self.delete_calls.append({"ids": ids, "where": where})
+        if where is not None and ids is None:
+            raise AssertionError(
+                "delete(where=...) is the call that hangs; it must never be used"
+            )
+        for i in ids or []:
+            self.docs.pop(i, None)
+
+    def count(self):
+        return len(self.docs)
+
+    def upsert(self, **kwargs):
+        for i, m in zip(kwargs.get("ids", []), kwargs.get("metadatas", [])):
+            self.docs[i] = m.get("doc_id")
+
+
+class _HangingWhereDeleteCollection(_RecordingCollection):
+    """delete(where=...) blocks forever, exactly like chromadb 1.5.6 did."""
+
+    def __init__(self, docs=None):
+        super().__init__(docs)
+        self.released = threading.Event()
+        self.where_delete_entered = threading.Event()
+
+    def delete(self, ids=None, where=None, **kwargs):
+        self.delete_calls.append({"ids": ids, "where": where})
+        if where is not None and ids is None:
+            self.where_delete_entered.set()
+            # Never returns unless the test explicitly lets go.
+            self.released.wait()
+            return
+        for i in ids or []:
+            self.docs.pop(i, None)
+
+
+class _FakeCodeIndex:
+    def __init__(self, chunks):
+        self.chunks = chunks
+
+    def _load_index(self):
+        pass
+
+
+def _make_index(tmp_path, collection) -> EmbeddingIndex:
+    """An EmbeddingIndex wired to *collection* with backends stubbed ready."""
+    idx = EmbeddingIndex(str(tmp_path), ollama_client=None)
+    idx._initialized = True
+    idx._available = True
+    idx._ollama_up = True
+    idx._model_ok = True
+    idx._ollama_ok = True
+    idx._collection = collection
+    return idx
+
+
+def _chunks_for(doc_id, n=3):
+    return {
+        f"{doc_id}::c{i}": {
+            "doc_id": doc_id,
+            "content": f"def sym{i}():\n    return {i}  # padding to clear the 20-char floor",
+            "name": f"sym{i}",
+            "type": "function",
+            "line_start": i * 10,
+            "line_end": i * 10 + 5,
+        }
+        for i in range(n)
+    }
+
+
+# ---------------------------------------------------------------------------
+# Bug 1 -- chunk removal must not use the hanging where-delete
+# ---------------------------------------------------------------------------
+
+
+def test_remove_file_chunks_deletes_by_id_never_by_where(tmp_path):
+    """The get-ids-then-delete-by-ids path is primary, not a fallback."""
+    col = _RecordingCollection({
+        "a.py::c0": "a.py", "a.py::c1": "a.py", "b.py::c0": "b.py",
+    })
+    idx = _make_index(tmp_path, col)
+
+    idx._remove_file_chunks("a.py")
+
+    # The file's chunks are gone, the other file's are untouched.
+    assert col.docs == {"b.py::c0": "b.py"}
+
+    # Every delete resolved explicit ids; none passed a where filter.
+    assert col.delete_calls, "expected at least one delete call"
+    for call in col.delete_calls:
+        assert call["where"] is None, f"where-delete reintroduced: {call}"
+        assert call["ids"] is not None
+
+    # The metadata filter moved to get(), which does return.
+    assert col.get_calls[0]["where"] == {"doc_id": "a.py"}
+
+
+def test_remove_file_chunks_returns_when_where_delete_would_hang(tmp_path):
+    """A regression that reintroduces delete(where=) must fail, not hang.
+
+    The fake blocks forever on where-delete. We join with a bound in a daemon
+    thread, so a regression surfaces as a failed assertion rather than a
+    wedged pytest run.
+    """
+    col = _HangingWhereDeleteCollection({"a.py::c0": "a.py", "a.py::c1": "a.py"})
+    idx = _make_index(tmp_path, col)
+
+    done = threading.Event()
+
+    def _run():
+        idx._remove_file_chunks("a.py")
+        done.set()
+
+    t = threading.Thread(target=_run, daemon=True)
+    t.start()
+    finished = done.wait(timeout=10)
+
+    col.released.set()  # let any wedged thread go so it cannot leak
+
+    assert finished, (
+        "_remove_file_chunks blocked -- it is calling delete(where=...) again"
+    )
+    assert not col.where_delete_entered.is_set()
+    assert col.docs == {}
+
+
+def test_remove_file_chunks_survives_include_kwarg_rejection(tmp_path):
+    """Older chromadb (>=0.4.24 is supported) may reject include=[]."""
+
+    class _NoIncludeCollection(_RecordingCollection):
+        def get(self, where=None, include=None, **kwargs):
+            if include is not None:
+                raise TypeError("get() got an unexpected keyword argument 'include'")
+            return super().get(where=where, **kwargs)
+
+    col = _NoIncludeCollection({"a.py::c0": "a.py"})
+    idx = _make_index(tmp_path, col)
+
+    idx._remove_file_chunks("a.py")
+
+    assert col.docs == {}
+    assert col.delete_calls == [{"ids": ["a.py::c0"], "where": None}]
+
+
+def test_remove_file_chunks_noop_when_nothing_matches(tmp_path):
+    col = _RecordingCollection({"b.py::c0": "b.py"})
+    idx = _make_index(tmp_path, col)
+
+    idx._remove_file_chunks("a.py")
+
+    assert col.delete_calls == []
+    assert col.docs == {"b.py::c0": "b.py"}
+
+
+def test_build_removes_stale_file_chunks_without_where_delete(tmp_path):
+    """End-to-end: build() drops a deleted file's chunks by id."""
+    col = _RecordingCollection({"gone.py::c0": "gone.py"})
+    idx = _make_index(tmp_path, col)
+    idx._file_hashes = {"gone.py": "deadbeef"}
+
+    class _Ollama:
+        def embed_batch(self, texts, model=None):
+            return [[0.1, 0.2, 0.3] for _ in texts]
+
+    idx.ollama = _Ollama()
+
+    result = idx.build(_FakeCodeIndex(_chunks_for("kept.py")))
+
+    assert "gone.py" not in idx._file_hashes
+    assert "gone.py::c0" not in col.docs
+    assert result.get("chunks_embedded", 0) == 3
+    for call in col.delete_calls:
+        assert call["where"] is None, f"where-delete reintroduced: {call}"
+
+
+# ---------------------------------------------------------------------------
+# Bug 2 -- a busy build lock degrades, it does not block
+# ---------------------------------------------------------------------------
+
+
+def test_build_degrades_instead_of_blocking_on_busy_lock(tmp_path, monkeypatch):
+    """build() must give up on a held lock, not park behind it forever."""
+    monkeypatch.setattr(ei_mod, "_BUILD_LOCK_WAIT_SECONDS", 0.2)
+
+    col = _RecordingCollection()
+    idx = _make_index(tmp_path, col)
+
+    holder_may_release = threading.Event()
+    holder_has_lock = threading.Event()
+
+    def _hold():
+        with idx._lock:
+            holder_has_lock.set()
+            holder_may_release.wait(timeout=30)
+
+    holder = threading.Thread(target=_hold, daemon=True)
+    holder.start()
+    assert holder_has_lock.wait(timeout=5), "holder never acquired the lock"
+
+    result_box = {}
+    done = threading.Event()
+
+    def _build():
+        result_box["r"] = idx.build(_FakeCodeIndex(_chunks_for("a.py")))
+        done.set()
+
+    t = threading.Thread(target=_build, daemon=True)
+    t.start()
+    finished = done.wait(timeout=10)
+
+    holder_may_release.set()
+    holder.join(timeout=5)
+
+    assert finished, "build() blocked on a held lock instead of degrading"
+    result = result_box["r"]
+    assert result["degraded"] is True
+    assert result["available"] is True
+    assert result["chunks_embedded"] == 0
+    assert "busy" in result["error"].lower()
+    # Degraded returns keep the normal stats shape so callers using .get()
+    # with defaults (cli/c3.py, cli/hub_server.py) do not KeyError.
+    for key in ("files_processed", "files_skipped", "chunks_skipped",
+                "errors", "total_embedded"):
+        assert key in result
+
+
+def test_busy_lock_warns_once_per_instance(tmp_path, monkeypatch, caplog):
+    monkeypatch.setattr(ei_mod, "_BUILD_LOCK_WAIT_SECONDS", 0.05)
+    idx = _make_index(tmp_path, _RecordingCollection())
+
+    with caplog.at_level(logging.WARNING, logger="c3.embedding_index"):
+        with idx._lock:
+            assert idx._acquire_build_lock() is False
+            assert idx._acquire_build_lock() is False
+            assert idx._acquire_build_lock() is False
+
+    warnings = [r for r in caplog.records if r.levelno == logging.WARNING]
+    assert len(warnings) == 1, f"expected exactly one WARNING, got {len(warnings)}"
+    assert "lock" in warnings[0].getMessage().lower()
+
+
+def test_acquire_build_lock_releases_cleanly(tmp_path):
+    """The happy path still takes and gives back the lock."""
+    idx = _make_index(tmp_path, _RecordingCollection())
+
+    assert idx._acquire_build_lock(timeout=1.0) is True
+    idx._lock.release()
+    # Still acquirable afterwards -- no leak.
+    assert idx._acquire_build_lock(timeout=1.0) is True
+    idx._lock.release()
+
+
+def test_build_releases_lock_even_when_body_raises(tmp_path):
+    """A failure mid-build must not leave the lock held forever.
+
+    This is the property the ``with self._lock`` -> ``try/finally`` rewrite
+    has to preserve.
+    """
+    idx = _make_index(tmp_path, _RecordingCollection())
+    idx._file_hashes = {"gone.py": "deadbeef"}
+
+    def _boom(doc_id):
+        raise RuntimeError("backend exploded mid-build")
+
+    idx._remove_file_chunks = _boom
+
+    with pytest.raises(RuntimeError):
+        idx.build(_FakeCodeIndex(_chunks_for("a.py")))
+
+    assert idx._lock.acquire(timeout=1.0), "build() leaked the lock"
+    idx._lock.release()
+
+
+if __name__ == "__main__":
+    raise SystemExit(pytest.main([__file__, "-v"]))