fix(serve): resource governor for multi-instance system freezes

src/context_engine/cli.py

@@ -3426,6 +3426,14 @@ async def _run_serve(config) -> None:
     # gets the multiprocess path.
     os.environ.setdefault("CCE_EMBED_PARALLEL", "0")
 
+    # Cap ONNX Runtime threads per process (#139). With N concurrent
+    # cce-serve processes, uncapped threads (default = cpu_count) create
+    # thousands of OS threads competing for cores and page-cache.
+    from context_engine.resource_governor import (
+        cap_ort_threads, IdleTracker, ProjectIndexLock, is_memory_pressured,
+    )
+    cap_ort_threads(max_threads=getattr(config, "serve_max_ort_threads", None))
+
     from context_engine.storage.local_backend import LocalBackend
     from context_engine.indexer.embedder import Embedder
     from context_engine.retrieval.retriever import HybridRetriever
@@ -3452,6 +3460,15 @@ async def _run_serve(config) -> None:
         embedder=embedder, config=config,
     )
 
+    # Idle tracker — shuts down the server after prolonged inactivity (#139).
+    idle_tracker = IdleTracker(
+        timeout_minutes=getattr(config, "serve_idle_timeout_minutes", None),
+    )
+
+    # Per-project index lock — prevents N duplicate cce-serve processes from
+    # all indexing the same project simultaneously (#139).
+    index_lock = ProjectIndexLock(storage_base)
+
     chunk_count = backend._vector_store.count()
     import sys
 
@@ -3474,15 +3491,39 @@ async def _on_file_change(file_path: str):
                 await _reindex_queue.put(rel)
 
         async def _reindex_worker():
-            """Background task that processes re-index requests sequentially."""
+            """Background task that processes re-index requests sequentially.
+
+            Acquires a per-project file lock before indexing so duplicate
+            cce-serve processes for the same project don't all index at
+            once.  Backs off when the system is under memory pressure
+            (Linux PSI).  See #139.
+            """
             while True:
                 rel = await _reindex_queue.get()
                 _reindex_pending.discard(rel)
+                # Back off under memory pressure (#139).
+                if is_memory_pressured():
+                    _log.info(
+                        "Memory pressure detected; deferring re-index of %s",
+                        rel,
+                    )
+                    _reindex_queue.task_done()
+                    await asyncio.sleep(30)
+                    continue
+                if not index_lock.try_acquire():
+                    _log.debug(
+                        "Another process is indexing this project; "
+                        "skipping %s", rel,
+                    )
+                    _reindex_queue.task_done()
+                    continue
                 try:
                     await run_indexing(config, project_dir, target_path=rel)
                     _log.debug("Re-indexed: %s", rel)
                 except Exception as exc:
                     _log.warning("Watch re-index failed for %s: %s", rel, exc)
+                finally:
+                    index_lock.release()
                 _reindex_queue.task_done()
 
         watcher = FileWatcher(
@@ -3535,11 +3576,18 @@ async def _reindex_worker():
     except Exception as exc:
         _log.warning("Auto-prune worker failed to start: %s", exc)
 
+    # Inject idle tracker so every MCP tool call resets the idle timer (#139).
+    mcp._idle_tracker = idle_tracker
+
     watcher_label = " · live watcher active" if watcher else ""
     hook_label = f" · memory hooks :{hook_port}" if hook_port else ""
+    idle_label = (
+        f" · idle shutdown {idle_tracker.timeout_seconds // 60}m"
+        if idle_tracker.timeout_seconds > 0 else ""
+    )
     print(
         f"CCE ready · {project_name} · {chunk_count} chunks indexed"
-        f"{watcher_label}{hook_label}",
+        f"{watcher_label}{hook_label}{idle_label}",
         file=sys.stderr,
     )
 
@@ -3551,6 +3599,26 @@ async def _reindex_worker():
     serve_loop = asyncio.get_running_loop()
     mcp_task = asyncio.create_task(mcp.run_stdio())
 
+    # Idle-shutdown loop: periodically check if the server has been unused
+    # for longer than the configured timeout and initiate shutdown (#139).
+    async def _idle_watchdog():
+        while True:
+            await asyncio.sleep(60)
+            if idle_tracker.is_idle():
+                idle_min = int(idle_tracker.idle_seconds // 60)
+                _log.info(
+                    "No MCP activity for %d minutes; shutting down", idle_min,
+                )
+                print(
+                    f"CCE idle for {idle_min}m — shutting down "
+                    f"(set CCE_IDLE_TIMEOUT_MINUTES=0 to disable)",
+                    file=sys.stderr,
+                )
+                mcp_task.cancel()
+                return
+
+    idle_task = asyncio.create_task(_idle_watchdog())
+
     def _request_shutdown(signame: str) -> None:
         if not mcp_task.done():
             _log.info("Received %s, shutting down...", signame)
@@ -3588,6 +3656,7 @@ def _request_shutdown(signame: str) -> None:
         except asyncio.CancelledError:
             pass
     finally:
+        idle_task.cancel()
         for _sig in installed_signals:
             try:
                 serve_loop.remove_signal_handler(_sig)
@@ -3618,3 +3687,4 @@ def _request_shutdown(signame: str) -> None:
                 await hook_runner.cleanup()
             except Exception:
                 _log.warning("hook_runner cleanup failed", exc_info=True)
+        index_lock.release()

src/context_engine/config.py

@@ -79,6 +79,12 @@ class Config:
     indexer_watch: bool = True
     indexer_debounce_ms: int = 500
     indexer_ignore: list[str] = field(default_factory=lambda: list(DEFAULT_IGNORE))
+    # Resource governor (#139) — caps per-process ONNX Runtime threads and
+    # auto-shuts down idle servers so zombie processes don't accumulate.
+    # 0 = disabled (no auto-shutdown / use ORT default threads).
+    serve_idle_timeout_minutes: int = 30
+    serve_max_ort_threads: int = 2
+
     # When True, the indexer skips well-known credential filenames
     # (.env*, *.pem, secrets.yml, credentials.json, …) and redacts
     # AWS/GitHub/JWT/etc. patterns from the content of files it does
@@ -140,6 +146,8 @@ def _deep_merge(base: dict, override: dict) -> dict:
     "indexer_watch": bool,
     "indexer_debounce_ms": int,
     "indexer_ignore": list,
+    "serve_idle_timeout_minutes": int,
+    "serve_max_ort_threads": int,
     "indexer_redact_secrets": bool,
     "memory_redact_pii": bool,
     "audit_log_enabled": bool,
@@ -162,6 +170,8 @@ def _apply_dict_to_config(config: Config, data: dict) -> None:
         ("retrieval", "top_k"): "retrieval_top_k",
         ("retrieval", "marginal_ratio"): "retrieval_marginal_ratio",
         ("retrieval", "bootstrap_max_tokens"): "bootstrap_max_tokens",
+        ("serve", "idle_timeout_minutes"): "serve_idle_timeout_minutes",
+        ("serve", "max_ort_threads"): "serve_max_ort_threads",
         ("indexer", "watch"): "indexer_watch",
         ("indexer", "debounce_ms"): "indexer_debounce_ms",
         ("indexer", "ignore"): "indexer_ignore",

src/context_engine/integration/mcp_server.py

@@ -389,6 +389,9 @@ def __init__(self, retriever, backend, compressor, embedder, config) -> None:
         self._compressor = compressor
         self._embedder = embedder
         self._config = config
+        # Set by _run_serve after construction; reset on every tool call
+        # so the idle-shutdown watchdog knows the server is in use (#139).
+        self._idle_tracker = None
         # Propagate the PII-redaction toggle to the memory module's
         # process-global state. Done at MCPServer boot — the compressor
         # and migrate paths read from the same module-level flag.
@@ -863,6 +866,8 @@ async def list_tools():
         @self._server.call_tool()
         async def call_tool(name: str, arguments: dict):
             arguments = arguments or {}
+            if self._idle_tracker is not None:
+                self._idle_tracker.touch()
             try:
                 if name == "context_search":
                     return await self._handle_context_search(arguments)

src/context_engine/resource_governor.py

@@ -0,0 +1,203 @@
+"""Resource governor — prevents multi-instance CCE from overwhelming the host.
+
+Addresses issue #139: dozens of ``cce serve`` processes (one per project per
+AI session) exhaust system memory and cause 10–20 minute desktop freezes on
+Linux.
+
+Three mechanisms:
+
+1. **ONNX Runtime thread caps** — each process gets a bounded number of
+   intra-op/inter-op threads instead of the OS default (often == CPU count).
+2. **Per-project index lock** — only one process indexes a given project at
+   a time; others skip and retry later.
+3. **Memory-pressure backoff** — on Linux, reads ``/proc/pressure/memory``
+   (PSI) and pauses heavy work when the system is under pressure. On other
+   platforms this is a no-op (always returns False).
+4. **Idle shutdown** — ``cce serve`` exits after N minutes of MCP inactivity
+   so zombie processes don't accumulate.
+"""
+
+from __future__ import annotations
+
+import logging
+import os
+import sys
+import time
+from pathlib import Path
+
+try:
+    import fcntl
+except ImportError:          # Windows
+    fcntl = None  # type: ignore[assignment]
+
+log = logging.getLogger(__name__)
+
+# ── 1. ONNX Runtime thread caps ───────────────────────────────────────────
+
+# How many threads each ONNX Runtime session (= one CCE embedding process)
+# may use.  With 68 CCE processes, uncapped threads (default = cpu_count)
+# create thousands of OS threads competing for cores and page-cache, which
+# is the primary amplifier of the OOM/freeze scenario.
+#
+# Env-var ``CCE_ORT_THREADS`` overrides; 0 means "use ONNX default".
+_DEFAULT_ORT_THREADS = 2
+
+
+def cap_ort_threads(max_threads: int | None = None) -> int:
+    """Set ONNX Runtime thread-count env vars *before* any model is loaded.
+
+    Must be called before ``fastembed.TextEmbedding()`` or any other ONNX
+    import.  Returns the value that was applied.
+    """
+    raw = os.environ.get("CCE_ORT_THREADS", "").strip()
+    if raw:
+        try:
+            n = int(raw)
+        except ValueError:
+            n = _DEFAULT_ORT_THREADS
+    elif max_threads is not None:
+        n = max_threads
+    else:
+        n = _DEFAULT_ORT_THREADS
+
+    if n <= 0:
+        return 0  # 0 = "let ONNX pick"
+
+    for var in (
+        "OMP_NUM_THREADS",
+        "MKL_NUM_THREADS",
+        "OPENBLAS_NUM_THREADS",
+        "ORT_NUM_THREADS",       # some ORT builds read this directly
+    ):
+        os.environ.setdefault(var, str(n))
+
+    # tokenizers library (used by fastembed) also spawns threads
+    os.environ.setdefault("TOKENIZERS_PARALLELISM", "false")
+
+    log.debug("ORT thread cap applied: %d", n)
+    return n
+
+
+# ── 2. Per-project index lock ──────────────────────────────────────────────
+
+class ProjectIndexLock:
+    """File-lock so only one ``cce serve`` process indexes a given project.
+
+    Usage::
+
+        lock = ProjectIndexLock(storage_base)
+        if lock.try_acquire():
+            try:
+                await run_indexing(...)
+            finally:
+                lock.release()
+        else:
+            log.debug("Another process is indexing; skipping this cycle")
+
+    The lock file lives at ``{storage_base}/.index.lock``.  ``fcntl.flock``
+    is advisory but sufficient — all competing processes are CCE.
+    """
+
+    def __init__(self, storage_base: Path) -> None:
+        self._lock_path = storage_base / ".index.lock"
+        self._fd: int | None = None
+
+    def try_acquire(self) -> bool:
+        """Non-blocking attempt. Returns True if this process now holds it."""
+        if fcntl is None:
+            return True  # no-op on Windows
+        try:
+            self._lock_path.parent.mkdir(parents=True, exist_ok=True)
+            fd = os.open(str(self._lock_path), os.O_CREAT | os.O_RDWR, 0o644)
+            fcntl.flock(fd, fcntl.LOCK_EX | fcntl.LOCK_NB)
+            self._fd = fd
+            return True
+        except (OSError, BlockingIOError):
+            return False
+
+    def release(self) -> None:
+        if self._fd is not None:
+            try:
+                if fcntl is not None:
+                    fcntl.flock(self._fd, fcntl.LOCK_UN)
+                os.close(self._fd)
+            except OSError:
+                pass
+            self._fd = None
+
+    def __enter__(self):
+        return self.try_acquire()
+
+    def __exit__(self, *_exc):
+        self.release()
+
+
+# ── 3. Memory-pressure detection (Linux PSI) ──────────────────────────────
+
+_PSI_PATH = Path("/proc/pressure/memory")
+# ``some avg10`` above this threshold means the system is under meaningful
+# memory pressure.  85% was the peak observed in the issue report; we use
+# a much lower trip-wire so CCE backs off before things get critical.
+_PSI_THRESHOLD_PCT = 25.0
+
+
+def is_memory_pressured() -> bool:
+    """Return True when the host is under significant memory pressure.
+
+    On Linux, reads PSI (Pressure Stall Information).  On other platforms
+    returns False (no-op).
+    """
+    if sys.platform != "linux":
+        return False
+    try:
+        text = _PSI_PATH.read_text()
+        # Format: "some avg10=X.XX avg60=X.XX avg300=X.XX total=N"
+        for line in text.splitlines():
+            if line.startswith("some "):
+                for part in line.split():
+                    if part.startswith("avg10="):
+                        val = float(part.split("=", 1)[1])
+                        if val > _PSI_THRESHOLD_PCT:
+                            return True
+                break
+    except Exception:
+        pass
+    return False
+
+
+# ── 4. Idle-timeout tracker ───────────────────────────────────────────────
+
+_DEFAULT_IDLE_TIMEOUT_MINUTES = 30
+
+
+class IdleTracker:
+    """Track MCP activity and signal when the server has been idle too long.
+
+    ``touch()`` is called on every MCP tool invocation.  ``is_idle()``
+    returns True when no tool has been called for ``timeout_minutes``.
+    """
+
+    def __init__(self, timeout_minutes: int | None = None) -> None:
+        raw = os.environ.get("CCE_IDLE_TIMEOUT_MINUTES", "").strip()
+        if raw:
+            try:
+                timeout_minutes = int(raw)
+            except ValueError:
+                pass
+        if timeout_minutes is None:
+            timeout_minutes = _DEFAULT_IDLE_TIMEOUT_MINUTES
+        # 0 or negative = disabled
+        self.timeout_seconds = timeout_minutes * 60 if timeout_minutes > 0 else 0
+        self._last_activity = time.monotonic()
+
+    def touch(self) -> None:
+        self._last_activity = time.monotonic()
+
+    def is_idle(self) -> bool:
+        if self.timeout_seconds <= 0:
+            return False
+        return (time.monotonic() - self._last_activity) > self.timeout_seconds
+
+    @property
+    def idle_seconds(self) -> float:
+        return time.monotonic() - self._last_activity