From 8c00040dd8ad1820c213dac92b6966e13da1b340 Mon Sep 17 00:00:00 2001 From: rajkumarsakthivel Date: Fri, 3 Jul 2026 21:30:41 +0100 Subject: [PATCH 1/2] docs: add C# to AST-aware supported languages in README --- README.md | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) diff --git a/README.md b/README.md index f979216..8311cff 100644 --- a/README.md +++ b/README.md @@ -391,7 +391,7 @@ No GPU required. With Ollama, embeddings are handled by the Ollama server. With ## Supported Languages -**AST-aware chunking (tree-sitter parsed, 10 extensions):** +**AST-aware chunking (tree-sitter parsed, 11 extensions):** | Language | Extensions | |----------|-----------| @@ -402,13 +402,14 @@ No GPU required. With Ollama, embeddings are handled by the Ollama server. With | Go | `.go` | | Rust | `.rs` | | Java | `.java` | +| C# | `.cs` | **Language-aware fallback chunking (40+ extensions):** | Category | Languages | |----------|-----------| | Web | HTML, CSS, SCSS, LESS, Vue, Svelte | -| Systems | C, C++, C#, Zig, Nim | +| Systems | C, C++, Zig, Nim | | Mobile | Swift, Kotlin, Dart | | Functional | Haskell, Scala, Clojure, Elixir, Erlang, F# | | Scripting | Ruby, Perl, Lua, R, Bash/Zsh | From 52b0271c83c85d38e583d541467477cf6ac95de4 Mon Sep 17 00:00:00 2001 From: rajkumarsakthivel Date: Fri, 24 Jul 2026 15:37:11 +0100 Subject: [PATCH 2/2] fix(serve): add resource governor to prevent multi-instance system freezes (#139) Multiple cce serve processes (one per project per AI session) exhaust system memory and create 10-20 minute desktop freezes. Root cause: no thread caps, no idle shutdown, no cross-process coordination, no memory pressure awareness. Adds resource_governor.py with four mechanisms: - ONNX Runtime thread caps (default 2 per process, via CCE_ORT_THREADS) - Per-project file lock so duplicate instances don't index simultaneously - Linux PSI memory-pressure detection to pause indexing under pressure - Idle auto-shutdown after 30 minutes of MCP inactivity New config keys: serve.idle_timeout_minutes, serve.max_ort_threads New env vars: CCE_ORT_THREADS, CCE_IDLE_TIMEOUT_MINUTES --- src/context_engine/cli.py | 74 ++++++- src/context_engine/config.py | 10 + src/context_engine/integration/mcp_server.py | 5 + src/context_engine/resource_governor.py | 203 +++++++++++++++++++ 4 files changed, 290 insertions(+), 2 deletions(-) create mode 100644 src/context_engine/resource_governor.py diff --git a/src/context_engine/cli.py b/src/context_engine/cli.py index 362ff12..63d2933 100644 --- a/src/context_engine/cli.py +++ b/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() diff --git a/src/context_engine/config.py b/src/context_engine/config.py index a4fc52b..63bd92e 100644 --- a/src/context_engine/config.py +++ b/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", diff --git a/src/context_engine/integration/mcp_server.py b/src/context_engine/integration/mcp_server.py index d359223..76691ed 100644 --- a/src/context_engine/integration/mcp_server.py +++ b/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) diff --git a/src/context_engine/resource_governor.py b/src/context_engine/resource_governor.py new file mode 100644 index 0000000..093e7c8 --- /dev/null +++ b/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