Function bodies 257 total

    def _is_connection_healthy(self) -> bool:
        """Check if existing connection is still valid.

        Returns:
            True if connection is healthy and can be reused, False otherwise.
        """
        if not self.sock or not self._identified:
            return False

        # Check idle timeout
        if self._last_activity > 0:
            idle_time = time.time() - self._last_activity
            if idle_time > MAX_IDLE_TIME:
                logger.debug(f"Connection idle for {idle_time:.1f}s, will reconnect")
                return False

        # Check if socket is still connected (non-blocking peek)
        try:
            self.sock.setblocking(False)
            try:
                data = self.sock.recv(1, socket.MSG_PEEK)
                if not data:
                    return False  # Connection closed by server
            except BlockingIOError:
                pass  # No data available = connection is alive
            finally:
                self.sock

    def send_command(
        self, method: str, params: dict[str, Any] | None = None, request_id: Any = None
    ) -> dict[str, Any]:
        """Send a JSON-RPC request to SketchUp and return the response.

        Args:
            method: The command/method name to invoke.
            params: Optional parameters for the command.
            request_id: Optional request ID for JSON-RPC.

        Returns:
            The result from the JSON-RPC response.

        Raises:
            SketchUpConnectionError: If connection fails or is lost.
            SketchUpProtocolError: If response is invalid JSON.
            SketchUpTimeoutError: If socket operation times out.
        """
        # Generate request_id if not provided
        if request_id is None:
            request_id = _next_request_id()

        # Reuse existing connection if healthy
        if not self._is_connection_healthy() and not self.connect():
            raise SketchUpConnectionError("Not connected to SketchUp")
   

def get_sketchup_connection(
    host: str = DEFAULT_HOST, port: int = DEFAULT_PORT, agent: str = "unknown"
) -> SketchupConnection:
    """Get or create a persistent SketchUp connection.

    Thread-safe singleton pattern for connection management.

    Args:
        host: Host to connect to.
        port: Port to connect to.
        agent: Agent identifier (e.g., "user", "Claude", "mcp").

    Returns:
        A SketchupConnection instance.
    """
    global _sketchup_connection, _connection_agent

    with _connection_lock:
        # If agent changed, recreate connection
        if _sketchup_connection is not None and _connection_agent != agent:
            logger.debug(f"Agent changed from {_connection_agent} to {agent}, recreating connection")
            with contextlib.suppress(Exception):
                _sketchup_connection.disconnect()
            _sketchup_connection = None

        if _sketchup_connection is not None:
            try:
                # Test connection heal

def __getattr__(name: str):
    """Lazy import to avoid loading server module at package import time.

    This prevents logging configuration from running when importing
    submodules like supex_driver.mcp.resources.
    """
    if name in ("mcp", "main"):
        from supex_driver.mcp.server import main, mcp
        if name == "mcp":
            return mcp
        return main
    raise AttributeError(f"module {__name__!r} has no attribute {name!r}")

def get_agent_name(ctx: McpContext | None = None) -> str:
    """Get agent name from MCP client info or environment.

    Priority:
    1. MCP clientInfo.name (from session.client_params.clientInfo)
    2. SUPEX_AGENT environment variable
    3. Default to "mcp"
    """
    global _mcp_client_name

    # Try to get from Context
    if ctx is not None:
        try:
            # Access: ctx.request_context.session.client_params.clientInfo.name
            request_context = getattr(ctx, 'request_context', None)
            if request_context:
                session = getattr(request_context, 'session', None)
                if session:
                    client_params = getattr(session, 'client_params', None)
                    if client_params:
                        client_info = getattr(client_params, 'clientInfo', None)
                        if client_info:
                            raw_name = getattr(client_info, 'name', None)
                            if raw_name and isin

def setup_logging() -> None:
    """Configure logging for the MCP server.

    Sets up file logging and configures the logging format.
    Only runs once, even if called multiple times.
    """
    global _logging_configured
    if _logging_configured:
        return
    _logging_configured = True

    # Setup file logging for stderr only (stdout is used by MCP protocol)
    log_dir = os.environ.get("SUPEX_LOG_DIR", os.path.expanduser("~/.supex/logs"))
    try:
        os.makedirs(log_dir, exist_ok=True)
        stderr_log_file = os.path.join(log_dir, "stderr.log")

        # Redirect stderr to log file while preserving original
        stderr_logger = open(stderr_log_file, 'a', encoding='utf-8')
        _log_files.append(stderr_logger)

        # Only tee stderr, never stdout (MCP protocol uses stdout)
        sys.stderr = TeeStream(sys.stderr, stderr_logger)
    except OSError:
        # If we can't create log directory, continue without file logging
        pass

    # Configure log

def call_tool(
    ctx: McpContext,
    method: str,
    params: dict[str, Any] | None = None,
    operation: str = "operation"
) -> str:
    """Execute a tool call with standardized error handling.

    Args:
        ctx: MCP request context
        method: Tool method name to call
        params: Optional parameters for the tool
        operation: Description for error logging

    Returns:
        JSON string with result or error information
    """
    try:
        sketchup = get_sketchup_connection(agent=get_agent_name(ctx))
        result = sketchup.send_command(
            method=method,
            params=params or {},
            request_id=ctx.request_id
        )
        return json.dumps(result)
    except (SketchUpConnectionError, SketchUpTimeoutError) as e:
        logger.error(f"Connection error during {operation}: {e}")
        return json.dumps({"success": False, "error": str(e), "error_type": "connection"})
    except SketchUpProtocolError as e:
        logger.error(

def check_sketchup_status(ctx: McpContext) -> str:
    """Check if SketchUp is connected and responding"""
    try:
        sketchup = get_sketchup_connection(agent=get_agent_name(ctx))
        result = sketchup.send_command(
            method="ping", params={}, request_id=ctx.request_id
        )
        return json.dumps(
            {
                "status": "connected",
                "version": result.get("version", "unknown"),
                "message": "SketchUp is connected and responding",
            }
        )
    except (SketchUpConnectionError, SketchUpTimeoutError) as e:
        return json.dumps(
            {
                "status": "disconnected",
                "error": str(e),
                "error_type": "connection",
                "message": "Make sure the SketchUp extension is running",
            }
        )
    except SketchUpProtocolError as e:
        return json.dumps(
            {
                "status": "error",
                "error": str(e

def export_scene(ctx: McpContext, format: str = "skp") -> str:
    """Export the current SketchUp scene

    Args:
        format: Export format (skp, obj, dae, stl, png, jpg)
    """
    return call_tool(ctx, "export_scene", {"format": format}, "export_scene")

def eval_ruby(ctx: McpContext, code: str) -> str:
    """Evaluate arbitrary Ruby code in SketchUp context

    Args:
        code: Ruby code to execute
    """
    try:
        logger.info(f"Evaluating Ruby code ({len(code)} characters)")

        sketchup = get_sketchup_connection(agent=get_agent_name(ctx))

        result = sketchup.send_command(
            method="eval_ruby", params={"code": code}, request_id=ctx.request_id
        )

        # Format response consistently
        response = {
            "success": True,
            "result": result.get("content", [{"text": "Success"}])[0].get(
                "text", "Success"
            )
            if isinstance(result.get("content"), list)
            and len(result.get("content", [])) > 0
            else result.get("result", "Success"),
        }

        return json.dumps(response)
    except (SketchUpConnectionError, SketchUpTimeoutError) as e:
        logger.error(f"Connection error evaluating Ruby code: {e}")
        r

def eval_ruby_file(ctx: McpContext, file_path: str) -> str:
    """Evaluate Ruby code from a file in SketchUp context

    Args:
        file_path: Absolute path to Ruby file to execute
    """
    logger.info(f"Evaluating Ruby file: {file_path}")
    return call_tool(ctx, "eval_ruby_file", {"file_path": file_path}, "eval_ruby_file")

def get_model_info(ctx: McpContext) -> str:
    """Get basic information about the current SketchUp model

    Returns model statistics including:
    - title: Model title
    - units: Current units (meters, feet, etc.)
    - num_faces: Number of faces in the model
    - num_edges: Number of edges
    - num_groups: Number of groups
    - num_components: Number of component instances
    - modified: Whether model has unsaved changes
    """
    return call_tool(ctx, "get_model_info", {}, "get_model_info")

def list_entities(ctx: McpContext, entity_type: str = "all") -> str:
    """List entities in the model

    Args:
        entity_type: Type to filter - one of: faces, edges, groups, components, all

    Returns list of entities with type, name, and layer information
    """
    return call_tool(ctx, "list_entities", {"entity_type": entity_type}, "list_entities")

def get_selection(ctx: McpContext) -> str:
    """Get currently selected entities in SketchUp

    Returns:
    - count: Number of selected entities
    - entities: List of selected entities with details (type, properties)
    """
    return call_tool(ctx, "get_selection", {}, "get_selection")

def get_layers(ctx: McpContext) -> str:
    """Get list of layers (tags) in the model

    Returns list of layers with name, visible state, and entity count
    """
    return call_tool(ctx, "get_layers", {}, "get_layers")

def get_materials(ctx: McpContext) -> str:
    """Get list of materials in the model

    Returns list of materials with name, color, and texture information
    """
    return call_tool(ctx, "get_materials", {}, "get_materials")

def get_camera_info(ctx: McpContext) -> str:
    """Get current camera position and settings

    Returns camera eye position, target, up vector, and field of view
    """
    return call_tool(ctx, "get_camera_info", {}, "get_camera_info")

def take_screenshot(
    ctx: McpContext,
    width: int = 1920,
    height: int = 1080,
    transparent: bool = False,
    output_path: str | None = None
) -> str:
    """Take a screenshot of the current SketchUp view and save to disk

    IMPORTANT: Returns file path only (not image data) to save context tokens!
    Use Read tool to view the screenshot if needed.

    Args:
        width: Image width in pixels (default 1920)
        height: Image height in pixels (default 1080)
        transparent: Whether to use transparent background (default False)
        output_path: Optional custom path for screenshot. If not provided,
                     saves to .tmp/screenshots/ with timestamp

    Returns:
        JSON with file_path where screenshot was saved (~200 tokens vs 21k!)
        Use Read tool on the file_path to view screenshot if necessary
    """
    params: dict[str, int | bool | str] = {
        "width": width,
        "height": height,
        "transparent": transparent
   

def take_batch_screenshots(
    ctx: McpContext,
    shots: list[dict[str, Any]],
    output_dir: str | None = None,
    base_name: str = "screenshot",
    width: int = 1920,
    height: int = 1080,
    transparent: bool = False,
    restore_camera: bool = True
) -> str:
    """Take multiple screenshots with different camera positions in a single batch.

    Designed for zero visual flicker - renders happen offscreen while preserving
    the user's current view. Camera is restored after batch completes.

    Args:
        shots: List of shot specifications. Each shot is a dict with:
            - camera: Camera specification dict with:
                - type: Camera type (default "standard_view"):
                    - "standard_view" with view: "top"|"front"|"right"|"left"|"back"|"bottom"|"iso"
                    - "custom" with eye: [x,y,z], target: [x,y,z], optional up/fov/perspective
                    - "zoom_entity" with entity_ids: [id1, id2, ...], optional padding
           

def open_model(ctx: McpContext, path: str) -> str:
    """Open a SketchUp model file

    Args:
        path: Absolute path to the .skp file to open

    Returns success status and model information
    """
    return call_tool(ctx, "open_model", {"path": path}, "open_model")

def save_model(ctx: McpContext, path: str | None = None) -> str:
    """Save the current SketchUp model

    Args:
        path: Optional absolute path to save to. If not provided, saves to current location

    Returns success status and saved file path
    """
    params = {"path": path} if path else {}
    return call_tool(ctx, "save_model", params, "save_model")

  def initialize(host, port)
    @host = host
    @port = port
    @socket = nil
    @request_id = 0
    @session = nil
  end

  def connect
    @socket = TCPSocket.new(@host, @port)
    response = send_hello
    @session = response.dig('result', 'session')
    response
  end

  def send_hello
    params = {
      pid: Process.pid,
      name: CLIENT_NAME,
      version: CLIENT_VERSION
    }
    params[:token] = AUTH_TOKEN if AUTH_TOKEN && !AUTH_TOKEN.empty?
    send_request('hello', params)
  end

  def send_request(method, params)
    raise 'Not connected' unless connected?

    @request_id += 1
    request = { jsonrpc: '2.0', method: method, id: @request_id, params: params }
    @socket.write("#{request.to_json}\n")
    @socket.flush
    read_response
  end

  def read_response
    line = @socket.gets
    raise IOError, 'Connection closed' unless line

    JSON.parse(line)
  rescue JSON::ParserError => e
    { 'error' => { 'message' => "Parse error: #{e.message}" } }
  end

  def connect_with_retry(max_retries: nil)
    retries = max_retries || Integer(ENV.fetch('SUPEX_REPL_RETRIES', DEFAULT_RETRIES))
    delay = INITIAL_DELAY

    (retries + 1).times do |attempt|
      return connect
    rescue Errno::ECONNREFUSED
      raise if attempt >= retries

      puts "Connection failed, retrying in #{delay.to_i}s... (#{attempt + 1}/#{retries})"
      sleep(delay)
      delay *= BACKOFF_MULTIPLIER
    end

  def reconnect
    close
    puts 'Connection lost. Reconnecting...'
    connect_with_retry
    puts "Reconnected! Session: #{@session}"
    true
  rescue Errno::ECONNREFUSED
    false
  end

  def eval_with_reconnect(code)
    self.eval(code)
  rescue IOError, Errno::ECONNRESET, Errno::EPIPE
    return { 'error' => { 'message' => 'Reconnection failed' } } unless reconnect

    self.eval(code)
  end

def server_available?(host, port)
  client = REPLClient.new(host, port)
  response = client.connect
  client.close
  !response['error']
rescue StandardError
  false
end

def run_simple_repl(host, port)
  puts 'Supex REPL Client - Simple Mode (JSON-RPC)'
  puts "Connecting to #{host}:#{port}..."

  client = REPLClient.new(host, port)
  begin
    response = client.connect_with_retry
    if response['error']
      puts "Error: #{response['error']['message']}"
      exit 1
    end

  def initialize(pry_instance)
    @pry = pry_instance
    @buffer = []
    @mutex = Mutex.new
    @timer_thread = nil
  end

  def add(line, options = {})
    result = classify_line(line)
    return result unless result == :buffered

    @mutex.synchronize do
      # Strip trailing newline but preserve the content
      @buffer << { line: line.to_s.chomp, options: options }
      start_flush_timer
    end

  def classify_line(line)
    return :ignore if line.nil?

    stripped = line.to_s.strip
    return :ignore if stripped.empty?

    return :bypass if PRY_COMMAND_PREFIXES.any? { |cmd| stripped.start_with?(cmd) }

    :buffered
  end

  def start_flush_timer
    @timer_thread&.kill
    @timer_thread = Thread.new do
      sleep(BUFFER_TIMEOUT_MS / 1000.0)
      @mutex.synchronize { do_flush }
    end

  def do_flush
    return if @buffer.empty?

    combined = @buffer.map { |item| item[:line] }.join("\n")
    options = @buffer.first[:options]
    @buffer.clear

    # Execute in separate thread to avoid deadlock with mutex
    Thread.new { @pry.supex_original_eval(combined, options) }.join
  end

def apply_pry_patch(host, port)
  $supex_client = REPLClient.new(host, port)

  begin
    response = $supex_client.connect_with_retry
    if response['error']
      puts "Supex REPL: Connection error: #{response['error']['message']}"
      return false
    end

    def eval(line, options = {})
      case supex_input_buffer.add(line, options)
      when :buffered
        true # Buffered - tell Pry to continue REPL loop
      when :ignore
        true # Empty line - ignore but continue REPL loop
      else # :bypass
        supex_original_eval(line, options) # Pry command - bypass buffering
      end

    def evaluate_ruby(code)
      response = $supex_client.eval_with_reconnect(code)
      if response['error']
        output.puts "Error: #{response['error']['message']}"
      else
        out = response.dig('result', 'output')
        output.print(out) if out
      end

def run_pry_repl(host, port)
  begin
    require 'pry'
  rescue LoadError
    puts 'Error: Pry gem not found. Install it with: gem install pry'
    exit 1
  end

      def execute(params)
        model = Sketchup.active_model
        return { success: false, error: 'No active model' } unless model

        shots = params['shots'] || []
        return { success: false, error: 'No shots specified' } if shots.empty?

        view = model.active_view
        output_dir = prepare_output_dir(params['output_dir'])
        base_name = params['base_name'] || 'screenshot'
        defaults = extract_defaults(params)

        # Save original camera state for restoration
        original_camera = save_camera_state(view.camera)

        # Process all shots with UI suppression for zero-flicker
        results = process_shots_with_ui_suppression(
          model, view, shots, output_dir, base_name, defaults
        )

        # Restore original camera if requested (default: true)
        restore_camera_state(view, original_camera) if params['restore_camera'] != false

        build_response(results, output_dir)
      rescue StandardError => e
        { success

      def process_shots_with_ui_suppression(model, view, shots, output_dir, base_name, defaults)
        results = []

        # Use start_operation with disable_ui=true to suppress UI updates
        # This minimizes flicker during rapid camera changes
        model.start_operation('Batch Screenshot', true)

        begin
          shots.each_with_index do |shot, index|
            result = process_single_shot(view, model, shot, output_dir, base_name, index, defaults)
            results << result
          end

      def process_single_shot(view, model, shot, output_dir, base_name, index, defaults)
        camera_spec = shot['camera'] || {}
        shot_name = shot['name'] || format('%03d', index)
        isolate_id = shot['isolate']

        width = shot['width'] || defaults[:width]
        height = shot['height'] || defaults[:height]

        isolation_state = nil

        begin
          # Apply isolation if requested (opens entity and hides rest of model)
          if isolate_id
            isolation_state = save_isolation_state(model)
            apply_isolation(model, isolate_id)
          end

      def apply_camera(view, model, camera_spec)
        type = camera_spec['type'] || 'standard_view'
        zoom = camera_spec['zoom_extents'] != false # Default true

        case type
        when 'standard_view'
          apply_standard_view(view, model, camera_spec['view'])
        when 'custom'
          apply_custom_camera(view, camera_spec)
        when 'zoom_entity'
          apply_zoom_entity(view, model, camera_spec)
          return # zoom_entity has its own zoom logic
        else
          raise "Unknown camera type: #{type}"
        end

      def apply_standard_view(view, model, view_name)
        config = STANDARD_VIEWS[view_name.to_s.downcase]
        raise "Unknown standard view: #{view_name}" unless config

        bounds = model.bounds
        center = bounds.empty? ? ORIGIN : bounds.center

        direction = Geom::Vector3d.new(*config[:direction]).normalize
        up = Geom::Vector3d.new(*config[:up])

        # Set arbitrary distance - zoom_extents will adjust if enabled
        eye = center.offset(direction.reverse, 100)

        camera = Sketchup::Camera.new(eye, center, up)
        camera.perspective = false # Standard views use parallel projection
        view.camera = camera
      end

      def apply_custom_camera(view, camera_spec)
        eye = Geom::Point3d.new(*camera_spec['eye'])
        target = Geom::Point3d.new(*camera_spec['target'])

        # Calculate view direction to check for parallel vectors
        view_direction = eye.vector_to(target)

        # Determine up vector - handle parallel case for top/bottom views
        default_up = Geom::Vector3d.new(0, 0, 1)
        up = if camera_spec['up']
               Geom::Vector3d.new(*camera_spec['up'])
             elsif view_direction.parallel?(default_up)
               # Looking straight up or down - use Y axis as up
               Geom::Vector3d.new(0, 1, 0)
             else
               default_up
             end

      def apply_zoom_entity(view, model, camera_spec)
        entity_ids = camera_spec['entity_ids'] || []
        raise 'No entity_ids specified for zoom_entity' if entity_ids.empty?

        entities = entity_ids.map { |id| model.find_entity_by_id(id) }.compact
        raise "No valid entities found for IDs: #{entity_ids}" if entities.empty?

        # Zoom to the entities
        view.zoom(entities)

        # Apply padding if specified
        padding = camera_spec['padding'] || 1.0
        apply_zoom_padding(view, padding) if padding != 1.0
      end

      def apply_zoom_padding(view, padding)
        camera = view.camera
        if camera.perspective?
          # For perspective, move camera back
          direction = camera.direction
          eye = camera.eye
          target = camera.target
          distance = eye.distance(target)
          new_distance = distance * padding
          new_eye = target.offset(direction.reverse, new_distance)
          camera.set(new_eye, target, camera.up)
        else
          # For parallel projection, increase height
          camera.height = camera.height * padding
        end

      def save_isolation_state(model)
        {
          active_path: model.active_path,
          inactive_hidden: model.rendering_options['InactiveHidden'],
          instance_hidden: model.rendering_options['InstanceHidden']
        }
      end

      def apply_isolation(model, entity_id)
        entity = model.find_entity_by_id(entity_id)
        raise "Entity not found for isolation: #{entity_id}" unless entity

        # Entity must be a Group or ComponentInstance
        unless entity.is_a?(Sketchup::Group) || entity.is_a?(Sketchup::ComponentInstance)
          raise "Can only isolate Group or ComponentInstance, got: #{entity.class}"
        end