tux.utils.hot_reload ¶

@abstractmethod
def scan_dependencies(self, file_path: Path) -> set[str]:
    """Scan file for dependencies."""

@abstractmethod
def get_dependents(self, module_name: str) -> set[str]:
    """Get direct dependents of a module."""

@abstractmethod
def get_transitive_dependents(self, module_name: str) -> set[str]:
    """Get all transitive dependents of a module."""

@span("dependency.get_file_hash")
def get_file_hash(self, file_path: Path) -> str:
    """Get SHA256 hash of file content for change detection."""
    try:
        with file_path.open("rb") as f:
            content = f.read()
            return hashlib.sha256(content).hexdigest()
    except OSError as e:
        logger.debug(f"Failed to read file {file_path}: {e}")
        return ""

def has_file_changed(self, file_path: Path, *, silent: bool = False) -> bool:
    """Check if file has actually changed since last scan."""
    current_hash = self.get_file_hash(file_path)
    file_key = str(file_path)

    if file_key not in self._file_hashes:
        # First time seeing this file - store hash but don't consider it "changed"
        # unless this is a brand new file that didn't exist before
        self._file_hashes[file_key] = current_hash
        if not silent:
            logger.debug(f"First time tracking {file_path.name}, storing baseline hash")
        return False  # Don't reload on first encounter

    if self._file_hashes[file_key] != current_hash:
        logger.debug(
            f"Content changed for {file_path.name}: hash {self._file_hashes[file_key][:8]} -> {current_hash[:8]}",
        )
        self._file_hashes[file_key] = current_hash
        return True

    logger.debug(f"No content change for {file_path.name}")
    return False

def clear_cache(self) -> None:
    """Clear the file hash cache."""
    self._file_hashes.clear()

@span("dependency.scan_classes")
def scan_class_definitions(self, file_path: Path, module_name: str) -> dict[str, dict[str, Any]]:
    """Scan for class definitions in a file for hot patching capabilities."""
    if not file_path.exists() or file_path.suffix != ".py":
        return {}

    try:
        with file_path.open(encoding="utf-8") as f:
            content = f.read()

        tree = ast.parse(content, filename=str(file_path))
        classes: dict[str, dict[str, Any]] = {}

        for node in ast.walk(tree):
            if isinstance(node, ast.ClassDef):
                base_names: list[str] = []
                for base in node.bases:
                    if isinstance(base, ast.Name):
                        base_names.append(base.id)
                    elif isinstance(base, ast.Attribute):
                        base_names.append(ast.unparse(base))

                classes[node.name] = {
                    "bases": base_names,
                    "lineno": node.lineno,
                    "module": module_name,
                }

    except Exception as e:
        logger.debug(f"Error scanning class definitions in {file_path}: {e}")
        if sentry_sdk.is_initialized():
            sentry_sdk.capture_exception(e)
        return {}
    else:
        return classes

def register_classes(self, module_name: str, file_path: Path) -> None:
    """Register class definitions for a module for hot patching tracking."""
    if classes := self.scan_class_definitions(file_path, module_name):
        self._class_registry[module_name] = classes
        logger.debug(f"Registered {len(classes)} classes for {module_name}: {list(classes.keys())}")

def get_changed_classes(self, module_name: str, file_path: Path) -> list[str]:
    """Detect which classes have changed in a module."""
    old_classes = self._class_registry.get(module_name, {})
    new_classes = self.scan_class_definitions(file_path, module_name)

    changed_classes: list[str] = []

    # Check for new or modified classes
    changed_classes.extend(
        class_name
        for class_name, class_info in new_classes.items()
        if class_name not in old_classes or old_classes[class_name] != class_info
    )
    # Check for removed classes
    changed_classes.extend(class_name for class_name in old_classes if class_name not in new_classes)

    # Update registry
    if new_classes:
        self._class_registry[module_name] = new_classes
    elif module_name in self._class_registry:
        del self._class_registry[module_name]

    return changed_classes

def clear_cache(self) -> None:
    """Clear the class registry cache."""
    self._class_registry.clear()

@span("dependency.scan_dependencies")
def scan_dependencies(self, file_path: Path) -> set[str]:
    """Scan a Python file for import dependencies."""
    if not file_path.exists() or file_path.suffix != ".py":
        return set()

    try:
        with file_path.open(encoding="utf-8") as f:
            content = f.read()

        tree = ast.parse(content, filename=str(file_path))
        dependencies: set[str] = set()

        for node in ast.walk(tree):
            if isinstance(node, ast.Import):
                self._process_import_node(node, dependencies)
            elif isinstance(node, ast.ImportFrom):
                self._process_import_from_node(node, dependencies, file_path)

    except Exception as e:
        logger.debug(f"Error scanning dependencies in {file_path}: {e}")
        if sentry_sdk.is_initialized():
            sentry_sdk.capture_exception(e)
        return set()
    else:
        return dependencies

def _process_import_node(self, node: ast.Import, dependencies: set[str]) -> None:
    """Process a regular import node."""
    for alias in node.names:
        if alias.name and alias.name.startswith(("tux.", "discord")):
            dependencies.add(alias.name)

def _process_import_from_node(self, node: ast.ImportFrom, dependencies: set[str], file_path: Path) -> None:
    """Process an import-from node."""
    if node.module and node.module.startswith(("tux.", "discord")):
        dependencies.add(node.module)
    elif node.level > 0:
        self._process_relative_import(node, dependencies, file_path)

def _process_relative_import(self, node: ast.ImportFrom, dependencies: set[str], file_path: Path) -> None:
    """Process relative imports."""
    if node.module:
        # Standard relative import: from .module import something
        if (
            abs_module := self._resolve_relative_import(file_path, node.module, node.level)
        ) and abs_module.startswith("tux."):
            dependencies.add(abs_module)
    else:
        # Pure relative import: from . import something
        for alias in node.names:
            if (
                alias.name
                and (abs_module := self._resolve_relative_import(file_path, None, node.level, alias.name))
                and abs_module.startswith("tux.")
            ):
                dependencies.add(abs_module)

def has_file_changed(self, file_path: Path, *, silent: bool = False) -> bool:
    """Check if file has actually changed since last scan."""
    return self._file_tracker.has_file_changed(file_path, silent=silent)

def register_classes(self, module_name: str, file_path: Path) -> None:
    """Register class definitions for a module for hot patching tracking."""
    if self._class_tracker:
        self._class_tracker.register_classes(module_name, file_path)

def get_changed_classes(self, module_name: str, file_path: Path) -> list[str]:
    """Detect which classes have changed in a module."""
    if self._class_tracker:
        return self._class_tracker.get_changed_classes(module_name, file_path)
    return []

def _resolve_relative_import(
    self,
    file_path: Path,
    module: str | None,
    level: int,
    imported_name: str | None = None,
) -> str | None:
    """Resolve relative imports to absolute module names.

    If `module` is None (pure relative import), treat as importing from the current package.
    """
    try:
        # Get the module path relative to tux package
        base_dir = Path(__file__).parent.parent
        relative_path = file_path.relative_to(base_dir)

        # Calculate the parent directory based on level
        path_parts = list(relative_path.parts[:-1])  # Remove filename

        # Go up 'level' directories
        for _ in range(level - 1):
            if path_parts:
                path_parts.pop()

        if module is None and imported_name is not None:
            # Pure relative import: from . import foo
            # Remove the last component (the module itself) to get the package
            package_parts = path_parts.copy()
            if package_parts:
                return f"tux.{'.'.join(package_parts)}.{imported_name}"
            return f"tux.{imported_name}"

        # Add the relative module if provided
        if module:
            path_parts.extend(module.split("."))

        if path_parts:
            return f"tux.{'.'.join(path_parts)}"
    except (ValueError, IndexError) as e:
        logger.debug(f"Failed to resolve relative import: {e}")

    return None

@span("dependency.update")
def update_dependencies(self, file_path: Path, module_name: str) -> None:
    """Update dependency tracking for a module."""
    if not self._config.enable_dependency_tracking:
        return

    dependencies = self.scan_dependencies(file_path)

    # Clean up old reverse dependencies
    if module_name in self._module_dependencies:
        for old_dep in self._module_dependencies[module_name]:
            if old_dep in self._reverse_dependencies:
                self._reverse_dependencies[old_dep].discard(module_name)
                if not self._reverse_dependencies[old_dep]:
                    del self._reverse_dependencies[old_dep]

    # Update forward dependencies
    self._module_dependencies[module_name] = dependencies

    # Update reverse dependencies
    for dep in dependencies:
        if dep not in self._reverse_dependencies:
            self._reverse_dependencies[dep] = set()
        self._reverse_dependencies[dep].add(module_name)

    # Register classes for hot patching
    self.register_classes(module_name, file_path)

    # Update scan time
    self._last_scan_time[module_name] = time.time()

    # Periodic cleanup
    self._cleanup_if_needed()

def get_dependents(self, module_name: str) -> set[str]:
    """Get direct dependents of a module."""
    return self._reverse_dependencies.get(module_name, set()).copy()

@span("dependency.get_transitive")
def get_transitive_dependents(self, module_name: str) -> set[str]:
    """Get all transitive dependents of a module with cycle detection."""
    visited: set[str] = set()
    result: set[str] = set()
    max_depth = self._config.max_dependency_depth

    def _visit(current_module: str, depth: int) -> None:
        if depth >= max_depth or current_module in visited:
            return

        visited.add(current_module)
        direct_dependents = self.get_dependents(current_module)

        for dependent in direct_dependents:
            if dependent not in result:
                result.add(dependent)
                _visit(dependent, depth + 1)

    _visit(module_name, 0)
    return result

def get_all_tracked_modules(self) -> list[str]:
    """Get all tracked modules."""
    return list(self._module_dependencies.keys())

def get_module_dependencies(self, module_name: str) -> set[str]:
    """Get direct dependencies of a module."""
    return self._module_dependencies.get(module_name, set()).copy()

def get_stats(self) -> dict[str, int]:
    """Get statistics about the dependency graph."""
    return {
        "total_modules": len(self._module_dependencies),
        "total_reverse_deps": len(self._reverse_dependencies),
        "cached_files": self._file_tracker.cache_size,
        "tracked_classes": self._class_tracker.tracked_classes_count if self._class_tracker else 0,
    }

def _cleanup_if_needed(self) -> None:
    """Perform cleanup if threshold is exceeded or enough time has passed."""
    current_time = time.time()

    should_cleanup = (
        self._file_tracker.cache_size > self._config.cleanup_threshold
        or current_time - self._last_cleanup > self._config.cache_cleanup_interval
    )

    if should_cleanup:
        self._cleanup_stale_entries()
        self._last_cleanup = current_time

def _cleanup_stale_entries(self) -> None:
    """Clean up stale entries from caches."""
    current_time = time.time()
    stale_threshold = 3600  # 1 hour

    # Clean up old scan times and associated data
    stale_modules = [
        module for module, scan_time in self._last_scan_time.items() if current_time - scan_time > stale_threshold
    ]

    for module in stale_modules:
        self._remove_module_tracking(module)

    if stale_modules:
        logger.debug(f"Cleaned up {len(stale_modules)} stale dependency entries")

def _remove_module_tracking(self, module_name: str) -> None:
    """Remove all tracking data for a module."""
    # Remove from scan times
    self._last_scan_time.pop(module_name, None)

    # Clean up dependencies
    if module_name in self._module_dependencies:
        for dep in self._module_dependencies[module_name]:
            if dep in self._reverse_dependencies:
                self._reverse_dependencies[dep].discard(module_name)
                if not self._reverse_dependencies[dep]:
                    del self._reverse_dependencies[dep]
        del self._module_dependencies[module_name]

    # Remove reverse dependencies
    if module_name in self._reverse_dependencies:
        del self._reverse_dependencies[module_name]

@span("dependency.hot_patch_class")
def hot_patch_class(self, module_name: str, class_name: str, new_class: type) -> bool:
    """Attempt to hot patch a class definition (experimental)."""
    if not self._config.enable_hot_patching:
        logger.debug("Hot patching disabled in configuration")
        return False

    try:
        if module_name not in sys.modules:
            logger.debug(f"Module {module_name} not loaded, cannot hot patch {class_name}")
            return False

        module = sys.modules[module_name]
        if not hasattr(module, class_name):
            logger.debug(f"Class {class_name} not found in {module_name}")
            return False

        # Attempt to patch
        setattr(module, class_name, new_class)
    except Exception as e:
        logger.error(f"Failed to hot patch class {class_name} in {module_name}: {e}")
        if sentry_sdk.is_initialized():
            sentry_sdk.capture_exception(e)
        return False
    else:
        logger.info(f"Hot patched class {class_name} in {module_name}")
        return True

@contextmanager
def cleanup_context(self):
    """Context manager for automatic cleanup."""
    try:
        yield self
    finally:
        self._file_tracker.clear_cache()
        if self._class_tracker:
            self._class_tracker.clear_cache()

@span("watcher.build_extension_map")
def _build_extension_map(self) -> None:
    """Build a map of file paths to extension names and scan initial dependencies."""
    extension_count = 0

    for extension in list(self.bot.extensions.keys()):
        if extension == "jishaku":
            continue

        try:
            path = path_from_extension(extension)
            if path.exists():
                self.path_to_extension[str(path)] = extension
                self.dependency_graph.update_dependencies(path, extension)
                extension_count += 1
            else:
                logger.warning(f"Could not find file for extension {extension}, expected at {path}")
        except Exception as e:
            logger.error(f"Error processing extension {extension}: {e}")
            if sentry_sdk.is_initialized():
                sentry_sdk.capture_exception(e)

    # Pre-populate hash cache for all Python files in watched directories
    # This eliminates "first encounter" issues for any file
    cached_files = self._populate_all_file_hashes()
    if cached_files > 0:
        logger.debug(f"Pre-populated hash cache for {cached_files} files")

    logger.debug(f"Mapped {extension_count} extensions for hot reload")

def _populate_all_file_hashes(self) -> int:
    """
    Pre-populate hash cache for all files in watched directories matching configured extensions.
    This can be disabled via configuration to avoid startup overhead.
    """
    if not self._config.prepopulate_hashes:
        logger.debug("Hash pre-population disabled in configuration")
        return 0

    cached_count = 0

    # Get the root watch path (this includes the entire tux directory)
    watch_root = Path(self.path)

    for ext in self._config.hash_extensions:
        for file_path in watch_root.rglob(f"*{ext}"):
            try:
                # Pre-populate cache silently using the public method
                self.dependency_graph.has_file_changed(file_path, silent=True)
                cached_count += 1
            except Exception as e:
                logger.warning(f"Failed to hash {file_path}: {e}")

    return cached_count

def start(self) -> None:
    """Start watching for file changes."""
    try:
        self.observer.start()
        logger.info(f"Hot reload watching {self.display_path}")
    except Exception as e:
        msg = f"Failed to start file watcher: {e}"
        raise FileWatchError(msg) from e

def stop(self) -> None:
    """Stop watching for file changes and cleanup resources."""
    try:
        self.observer.stop()
        self.observer.join(timeout=5.0)  # Add timeout to prevent hanging
        if self.observer.is_alive():
            logger.warning("File watcher observer thread did not terminate within the timeout period.")
    except Exception as e:
        logger.error(f"Error stopping file watcher: {e}")

    # Cancel any pending tasks
    for task in self.pending_tasks:
        if not task.done():
            task.cancel()

    # Cancel debounce timers
    for timer in self._debounce_timers.values():
        timer.cancel()
    self._debounce_timers.clear()

    logger.info("Stopped watching for changes")

@span("watcher.on_modified")
def on_modified(self, event: watchdog.events.FileSystemEvent) -> None:
    """Handle file modification events with debouncing and validation."""
    # Skip non-Python files and directories
    if event.is_directory or not str(event.src_path).endswith(".py"):
        return

    file_path = Path(str(event.src_path))

    # Skip temporary/backup files
    if file_path.name.startswith(".") or file_path.name.endswith((".tmp", ".bak", ".swp")):
        return

    logger.debug(f"File modification event: {file_path.name}")

    # Check if file actually changed - this prevents unnecessary reloads on save without changes
    if not self.dependency_graph.has_file_changed(file_path):
        logger.debug(f"File {file_path.name} saved but content unchanged, skipping reload")
        return

    logger.debug(f"File {file_path.name} content changed, scheduling reload")
    file_key = str(file_path)

    # Cancel existing debounce timer if any
    if file_key in self._debounce_timers:
        self._debounce_timers[file_key].cancel()

    # Set new debounce timer
    try:
        self._debounce_timers[file_key] = self.loop.call_later(
            self._config.debounce_delay,
            self._handle_file_change_debounced,
            file_path,
        )
    except Exception as e:
        logger.error(f"Failed to schedule file change handler: {e}")

def _handle_file_change_debounced(self, file_path: Path) -> None:
    """Handle file change after debounce period with comprehensive error handling."""
    file_key = str(file_path)

    # Remove from debounce tracking
    if file_key in self._debounce_timers:
        del self._debounce_timers[file_key]

    logger.debug(f"Processing file change: {file_path.name}")

    # Validate syntax before attempting reload (if enabled)
    if self._config.validate_syntax and file_path.suffix == ".py" and not validate_python_syntax(file_path):
        logger.debug(f"Skipping hot reload for {file_path.name} due to syntax errors")
        return

    try:
        # Handle special cases first
        if self._handle_special_files(file_path):
            return

        # Handle regular extension files
        self._handle_extension_file(file_path)
    except Exception as e:
        logger.error(f"Error handling file change for {file_path}: {e}")
        if sentry_sdk.is_initialized():
            sentry_sdk.capture_exception(e)