Debug-MCP

""" Runner subprocess entry point as a standalone script. This allows launching the runner with a specific Python interpreter via subprocess.Popen instead of multiprocessing.Process. Usage: python runner_main.py <workspace_root> Communication via JSON Lines on stdin/stdout. """ import json import sys from contextlib import redirect_stderr, redirect_stdout from io import StringIO from pathlib import Path # Ensure src is in path for imports sys.path.insert(0, str(Path(__file__).parent.parent)) from mcp_debug_tool.debugger import DebugController def _convert_paths_to_str(obj): """Recursively convert Path objects to strings.""" if isinstance(obj, Path): import os try: return os.fspath(obj) except (TypeError, AttributeError): try: if hasattr(obj, 'parts'): return '/'.join(obj.parts) if obj.parts else '' elif hasattr(obj, '__fspath__'): return obj.__fspath__() else: return repr(obj) except Exception: return '<invalid path object>' elif isinstance(obj, dict): return {k: _convert_paths_to_str(v) for k, v in obj.items()} elif isinstance(obj, list): return [_convert_paths_to_str(item) for item in obj] elif isinstance(obj, tuple): return tuple(_convert_paths_to_str(item) for item in obj) else: return obj class StdioDebugRunner: """Runner that communicates via JSON Lines on stdin/stdout.""" def __init__(self, workspace_root: Path): self.workspace_root = workspace_root self.debugger = DebugController() self._setup_python_path(workspace_root) # Save original stdout/stderr self._original_stdout = sys.stdout self._original_stderr = sys.stderr # Redirect print to stderr to keep stdout clean for JSON import builtins self._original_print = builtins.print builtins.print = lambda *args, **kwargs: self._original_print(*args, **kwargs, file=sys.stderr) def _setup_python_path(self, workspace_root: Path): """Setup Python path to include workspace's virtual environment.""" import os venv_paths = [ workspace_root / ".venv", workspace_root / "venv", workspace_root / ".env", ] for venv_path in venv_paths: if venv_path.exists(): lib_path = venv_path / "lib" if lib_path.exists(): for python_dir in lib_path.iterdir(): if python_dir.name.startswith("python"): potential_site = python_dir / "site-packages" if potential_site.exists(): site_packages_str = str(potential_site) if site_packages_str not in sys.path: sys.path.insert(0, site_packages_str) python_path = os.environ.get("PYTHONPATH", "") if python_path: os.environ["PYTHONPATH"] = f"{site_packages_str}:{python_path}" else: os.environ["PYTHONPATH"] = site_packages_str break break def run(self): """Main loop - read commands from stdin, write responses to stdout.""" try: for line in sys.stdin: line = line.strip() if not line: continue try: message = json.loads(line) command = message.get("command") if command == "run_to_breakpoint": self._handle_run_to_breakpoint(message) elif command == "continue": self._handle_continue(message) elif command == "terminate": break else: self._send_error(f"Unknown command: {command}") except json.JSONDecodeError as e: self._send_error(f"Invalid JSON: {e}") except Exception as e: self._send_error(f"Command error: {type(e).__name__}: {e}") except Exception as e: self._send_error(f"Runner error: {type(e).__name__}: {e}") def _handle_run_to_breakpoint(self, message: dict): """Handle run_to_breakpoint command.""" import os try: script_path = Path(message["script_path"]) file = message["file"] line = message["line"] args = message.get("args", []) env = message.get("env", {}) # Apply environment variables BEFORE any execution if env: os.environ.update(env) # Apply command-line arguments BEFORE any execution if args: sys.argv = [str(script_path)] + args # Lock working directory to workspace BEFORE any execution os.chdir(self.workspace_root) # PHASE 1: Compile and initialize with normal stdout (no capture yet) # This ensures imports and compilation happen with correct stdout/stderr if not self.debugger.globals_dict: try: # Read and compile with normal stdout with open(script_path) as f: code = f.read() compiled = compile(code, str(script_path), "exec") self.debugger.script_code = compiled self.debugger.script_path = script_path # Initialize globals self.debugger.globals_dict = { "__name__": "__main__", "__file__": str(script_path), } except Exception as e: # Compilation error - send immediately without capture self._send_error(f"Compilation error: {type(e).__name__}: {e}") return # Set breakpoint target self.debugger.breakpoint_hit = False self.debugger.current_frame = None self.debugger.current_locals = {} self.debugger.execution_completed = False self.debugger.execution_error = None self.debugger.target_breakpoint = (str(file), line) # Execute with stdout/stderr capture using controller helper captured_stdout = StringIO() captured_stderr = StringIO() with redirect_stdout(captured_stdout), redirect_stderr(captured_stderr): response = self.debugger.run_to_breakpoint(script_path, file, line) # Convert response to dict try: response_data = response.model_dump(mode='json') response_data = _convert_paths_to_str(response_data) # Add captured output to response stdout_content = captured_stdout.getvalue() stderr_content = captured_stderr.getvalue() if stdout_content: response_data["_captured_stdout"] = stdout_content if stderr_content: response_data["_captured_stderr"] = stderr_content except Exception as e: print(f"DEBUG: model_dump failed: {type(e).__name__}: {e}", file=self._original_stderr) raise # Send response self._send_response({"status": "success", "data": response_data}) except Exception as e: self._send_error(f"Breakpoint error: {type(e).__name__}: {e}") def _handle_continue(self, message: dict): """Handle continue command.""" try: file = message.get("file") line = message.get("line") if not file or not line: self._send_error("Continue requires file and line parameters") return if not self.debugger.globals_dict: self._send_error("No active execution context. Run to breakpoint first.") return captured_stdout = StringIO() captured_stderr = StringIO() with redirect_stdout(captured_stdout), redirect_stderr(captured_stderr): response, new_debugger = self.debugger.continue_execution(file, line) self.debugger = new_debugger # Convert response to dict try: response_data = response.model_dump(mode='json') response_data = _convert_paths_to_str(response_data) # Add captured output stdout_content = captured_stdout.getvalue() stderr_content = captured_stderr.getvalue() if stdout_content: response_data["_captured_stdout"] = stdout_content if stderr_content: response_data["_captured_stderr"] = stderr_content except Exception as e: print(f"DEBUG: model_dump failed in continue: {type(e).__name__}: {e}", file=self._original_stderr) raise self._send_response({"status": "success", "data": response_data}) except Exception as e: self._send_error(f"Continue error: {type(e).__name__}: {e}") def _send_response(self, response: dict): """Send response as JSON line to stdout.""" # Use original stdout to avoid interference json.dump(response, self._original_stdout) self._original_stdout.write('\n') self._original_stdout.flush() def _send_error(self, error_message: str): """Send error response to stdout.""" self._send_response({"status": "error", "message": error_message}) def main(): """Entry point for standalone runner.""" if len(sys.argv) < 2: print("Usage: python runner_main.py <workspace_root>", file=sys.stderr) sys.exit(1) workspace_root = Path(sys.argv[1]) runner = StdioDebugRunner(workspace_root) runner.run() if __name__ == "__main__": main()