import ctypes
import json
import platform
from pathlib import Path
from typing import Optional, Dict, Any, List, Tuple, Callable
# Global variable to hold the loaded library instance
# This avoids reloading the DLL on every call, which can be inefficient
# and problematic on some OSes if the library is already in use.
s_rust_lib: Optional[ctypes.CDLL] = None
s_lib_path: Optional[Path] = None
class FFIError(Exception):
"""Custom exception for FFI related errors."""
def __init__(self, message: str, details: Optional[Dict[str, Any]] = None):
super().__init__(message)
self.details = details if details is not None else {}
def _find_rust_library_path() -> Optional[Path]:
"""
Finds the Rust library path, checking for release and debug builds.
The path is constructed relative to this script's location and is OS-aware.
"""
base_path = Path(__file__).resolve(
).parent.parent # Moves up to the project root (d:/AIProjects/MCPServers/project-context-server)
system = platform.system()
if system == "Windows":
lib_name = "file_scanner.dll"
elif system == "Darwin": # macOS
lib_name = "libfile_scanner.dylib"
else: # Linux and other UNIX-like
lib_name = "libfile_scanner.so"
scanner_path = base_path / "file_scanner"
release_path = scanner_path / "target" / "release" / lib_name
if release_path.exists():
return release_path
debug_path = scanner_path / "target" / "debug" / lib_name
if debug_path.exists():
return debug_path
return None
def _get_rust_library() -> ctypes.CDLL:
"""
Loads the Rust library using ctypes.
Raises FFIError if the library cannot be found or loaded.
Uses a global variable to cache the loaded library.
"""
global s_rust_lib, s_lib_path
if s_rust_lib is not None and s_lib_path is not None and s_lib_path.exists():
# Potentially add a check here if the library file has been modified,
# though for simplicity, we assume it doesn't change during a single server run.
return s_rust_lib
s_lib_path = _find_rust_library_path()
if not s_lib_path:
raise FFIError("Rust library not found.", {
"tried_paths": "release and debug target directories"})
try:
s_rust_lib = ctypes.CDLL(str(s_lib_path))
# Setup free_string function once
s_rust_lib.free_string.argtypes = [ctypes.c_void_p]
s_rust_lib.free_string.restype = None
return s_rust_lib
except OSError as e:
s_rust_lib = None # Reset on failure
s_lib_path = None
raise FFIError(f"Failed to load Rust library: {e}", {
"path": str(s_lib_path)})
def _invoke_ffi_function(
rust_fn_name: str,
arg_types: List[Any],
args: Tuple[Any, ...],
debug: bool = False,
calling_function_name: str = "unknown"
) -> Dict[str, Any]:
"""
Generic helper to invoke a Rust FFI function that returns a JSON string.
Handles loading the library, setting up argtypes/restype, calling, and processing the response.
"""
ffi_debug_log: List[str] = []
if debug:
ffi_debug_log.append(
f"[_invoke_ffi_function for {calling_function_name}] Called. Rust func: {rust_fn_name}, Debug: {debug}")
try:
rust_lib = _get_rust_library()
rust_function = getattr(rust_lib, rust_fn_name)
rust_function.argtypes = arg_types
# All our Rust functions return char* (via void*)
rust_function.restype = ctypes.c_void_p
if debug:
# Be careful about logging sensitive data if args can contain it.
# For now, logging types and existence.
arg_summary = [(type(arg), arg.value if hasattr(arg, 'value') and isinstance(
arg.value, bytes) else '...') for arg in args]
ffi_debug_log.append(
f"[_invoke_ffi_function] Calling Rust '{rust_fn_name}' with arg types: {arg_summary}")
result_ptr = rust_function(*args)
if not result_ptr:
# Rust function returned a null pointer.
rust_lib.free_string(result_ptr) # type: ignore
error_msg = f"Rust function '{rust_fn_name}' returned a null pointer."
if debug:
ffi_debug_log.append(error_msg)
return {"error": error_msg, "debug_log": ffi_debug_log}
# Cast the void* to char*, get the value, and decode
value = ctypes.cast(result_ptr, ctypes.c_char_p).value
json_string = value.decode('utf-8') if value else ""
rust_lib.free_string(result_ptr) # type: ignore
if not json_string:
error_msg = f"Rust function '{rust_fn_name}' returned an empty string after decode."
if debug:
ffi_debug_log.append(error_msg)
return {"error": error_msg, "debug_log": ffi_debug_log}
if debug:
ffi_debug_log.append(
f"[_invoke_ffi_function] Raw JSON from '{rust_fn_name}': {json_string[:500]}...")
try:
result_data = json.loads(json_string)
if debug:
# Prepend FFI logs to any logs from Rust
rust_debug_logs = result_data.get("debug_log", [])
if not isinstance(rust_debug_logs, list):
rust_debug_logs = [
str(rust_debug_logs)] if rust_debug_logs is not None else []
result_data["debug_log"] = ffi_debug_log + rust_debug_logs
return result_data
except json.JSONDecodeError as e:
error_msg = f"Failed to parse JSON response from Rust function '{rust_fn_name}': {e}"
if debug:
ffi_debug_log.append(
f"{error_msg}. Raw string: {json_string[:500]}...")
return {"error": error_msg, "raw_response": json_string, "debug_log": ffi_debug_log}
except FFIError as e: # Errors from _get_rust_library
if debug:
ffi_debug_log.append(f"FFIError: {str(e)}. Details: {e.details}")
return {"error": str(e), "details": e.details, "debug_log": ffi_debug_log}
except AttributeError as e: # getattr failed for rust_fn_name
error_msg = f"Rust function '{rust_fn_name}' not found in library."
if debug:
ffi_debug_log.append(f"{error_msg} Details: {str(e)}")
return {"error": error_msg, "debug_log": ffi_debug_log}
except Exception as e:
# Catch any other unexpected errors
error_msg = f"An unexpected error occurred in _invoke_ffi_function for '{rust_fn_name}': {e}"
if debug:
ffi_debug_log.append(error_msg)
return {"error": error_msg, "debug_log": ffi_debug_log}
# --- Public FFI Invocation Functions ---
def invoke_scan_and_parse(
project_path: str, extensions: List[str], compactness_level: int, timeout_sec: int, debug: bool = False
) -> Dict[str, Any]:
"""
Invokes the 'scan_and_parse' FFI function.
"""
extensions_str = ",".join(extensions)
timeout_ms = timeout_sec * 1000
# Prepare ctype arguments
root_path_c = ctypes.c_char_p(project_path.encode('utf-8'))
extensions_c = ctypes.c_char_p(extensions_str.encode('utf-8'))
compactness_level_c = ctypes.c_uint8(compactness_level)
timeout_ms_c = ctypes.c_uint32(timeout_ms)
debug_c = ctypes.c_bool(debug)
arg_types = [ctypes.c_char_p, ctypes.c_char_p,
ctypes.c_uint8, ctypes.c_uint32, ctypes.c_bool]
args_tuple = (root_path_c, extensions_c,
compactness_level_c, timeout_ms_c, debug_c)
return _invoke_ffi_function("scan_and_parse", arg_types, args_tuple, debug, "invoke_scan_and_parse")
def invoke_project_wide_search(
project_path: str, search_string: str, extensions: List[str], context_lines: int, timeout_sec: int, debug: bool = False
) -> Dict[str, Any]:
"""
Invokes the 'project_wide_search' FFI function.
"""
extensions_str = ",".join(extensions)
timeout_ms = timeout_sec * 1000
root_path_c = ctypes.c_char_p(project_path.encode('utf-8'))
search_string_c = ctypes.c_char_p(search_string.encode('utf-8'))
extensions_c = ctypes.c_char_p(extensions_str.encode('utf-8'))
context_lines_c = ctypes.c_uint8(context_lines)
timeout_ms_c = ctypes.c_uint32(timeout_ms)
debug_c = ctypes.c_bool(debug)
arg_types = [ctypes.c_char_p, ctypes.c_char_p, ctypes.c_char_p,
ctypes.c_uint8, ctypes.c_uint32, ctypes.c_bool]
args_tuple = (root_path_c, search_string_c, extensions_c,
context_lines_c, timeout_ms_c, debug_c)
return _invoke_ffi_function("project_wide_search", arg_types, args_tuple, debug, "invoke_project_wide_search")
def invoke_concept_search(
project_path: str, query: str, extensions: List[str], top_n: int, timeout_sec: int, debug: bool = False
) -> Dict[str, Any]:
"""
Invokes the 'concept_search' FFI function.
Note: extensions are passed as a JSON string to Rust for concept_search.
"""
extensions_json_str = json.dumps(extensions)
timeout_ms = timeout_sec * 1000
root_path_c = ctypes.c_char_p(project_path.encode('utf-8'))
query_c = ctypes.c_char_p(query.encode('utf-8'))
extensions_json_c = ctypes.c_char_p(extensions_json_str.encode('utf-8'))
top_n_c = ctypes.c_size_t(top_n)
timeout_ms_c = ctypes.c_uint32(timeout_ms)
debug_c = ctypes.c_bool(debug)
arg_types = [ctypes.c_char_p, ctypes.c_char_p, ctypes.c_char_p,
ctypes.c_size_t, ctypes.c_uint32, ctypes.c_bool]
args_tuple = (root_path_c, query_c, extensions_json_c,
top_n_c, timeout_ms_c, debug_c)
# Special handling for concept_search results
raw_result = _invoke_ffi_function(
"concept_search", arg_types, args_tuple, debug, "invoke_concept_search")
# Ensure debug_log list exists if debug is true, done early.
if debug and "debug_log" not in raw_result:
# Should be created by _invoke_ffi_function if debug, but as a safeguard.
raw_result["debug_log"] = []
# Handle cases where Rust might return `{"error": null, "results": [...]}` on success.
# If "error" key exists and is None, and results are present, treat as success.
if raw_result.get("error") is None and "error" in raw_result and raw_result.get("results"):
if debug:
log_msg = "[invoke_concept_search] Corrected 'error: null' from Rust because results were present."
# Ensure debug_log is a list before trying to insert
if not isinstance(raw_result.get("debug_log"), list):
raw_result["debug_log"] = []
raw_result["debug_log"].insert(0, log_msg)
del raw_result["error"] # Remove the "error": null
if "status" not in raw_result: # If Rust didn't also provide a status
raw_result["status"] = "success" # Assume success
# Legacy handling for "error_adapter_call" status from Rust with string results.
# This might be less relevant if concept_search results are now consistently lists.
if "error" not in raw_result: # Check again, as the block above might have removed "error"
current_status_from_rust = raw_result.get("status")
results_data = raw_result.get("results") # Can be list or string
# This condition specifically checks for string results, as per original logic.
if current_status_from_rust == "error_adapter_call" and \
isinstance(results_data, str) and results_data.strip():
new_status = "success_with_rust_reported_issue"
raw_result["status"] = new_status
# 'error' key might be absent or was 'error:null' and removed.
# If Rust set status=error_adapter_call but no actual error field, add one.
if "error" not in raw_result:
raw_result["error"] = (
f"Rust layer reported status '{current_status_from_rust}' "
f"but provided results (type: {type(results_data).__name__})."
)
if debug:
log_message = (
f"[invoke_concept_search] Overrode Rust's status '{current_status_from_rust}' "
f"to '{new_status}' because results (string) were present. "
f"Ensured 'error' field: '{raw_result.get('error')}'."
)
if not isinstance(raw_result.get("debug_log"), list): # Should be a list
raw_result["debug_log"] = []
raw_result["debug_log"].insert(0, log_message)
return raw_result
# Example of how to potentially unload the library if needed, e.g., for testing or specific scenarios.
# This is OS-dependent and can be tricky.
# For Windows:
# if platform.system() == "Windows":
# kernel32 = ctypes.WinDLL('kernel32', use_last_error=True)
# def unload_library():
# global s_rust_lib
# if s_rust_lib:
# handle = s_rust_lib._handle
# kernel32.FreeLibrary(handle)
# s_rust_lib = None
# else: # For Linux/macOS (conceptual, dlclose is harder to call safely via ctypes)
# # On Unix, library unloading is typically handled by GC or less explicitly needed.
# # Forcing dlclose can be complex.
# def unload_library():
# global s_rust_lib
# # This is non-trivial and often not recommended with ctypes
# # Forcing GC might be an option: import gc; gc.collect()
# s_rust_lib = None # Allow GC to collect
# pass
# For now, we rely on Python's GC to unload the DLL when s_rust_lib is no longer referenced
# or when the program exits. Explicit unloading is commented out.