autocoder/temp_cleanup.py

"""
Temp Cleanup Module
===================

Cleans up stale temporary files and directories created by AutoForge agents,
Playwright, Node.js, and other development tools.

Called at Maestro (orchestrator) startup to prevent temp folder bloat.

Why this exists:
- Playwright creates browser profiles and artifacts in %TEMP%
- Node.js creates .node cache files (~7MB each, can accumulate to GBs)
- MongoDB Memory Server downloads binaries to temp
- These are never cleaned up automatically

When cleanup runs:
- At Maestro startup (when you click Play or auto-restart after rate limits)
- Only files/folders older than 1 hour are deleted (safe for running processes)
"""

import logging
import shutil
import tempfile
import time
from pathlib import Path

logger = logging.getLogger(__name__)

# Max age in seconds before a temp item is considered stale (1 hour)
MAX_AGE_SECONDS = 3600

# Directory patterns to clean up (glob patterns)
DIR_PATTERNS = [
    "playwright_firefoxdev_profile-*",  # Playwright Firefox profiles
    "playwright-artifacts-*",           # Playwright test artifacts
    "playwright-transform-cache",       # Playwright transform cache
    "mongodb-memory-server*",           # MongoDB Memory Server binaries
    "ng-*",                             # Angular CLI temp directories
    "scoped_dir*",                      # Chrome/Chromium temp directories
    "node-compile-cache",               # Node.js V8 compile cache directory
]

# File patterns to clean up (glob patterns)
FILE_PATTERNS = [
    ".[0-9a-f]*.node",   # Node.js/V8 compile cache files (~7MB each, varying hex prefixes)
    "claude-*-cwd",   # Claude CLI working directory temp files
    "mat-debug-*.log",  # Material/Angular debug logs
]


def cleanup_stale_temp(max_age_seconds: int = MAX_AGE_SECONDS) -> dict:
    """
    Clean up stale temporary files and directories.

    Only deletes items older than max_age_seconds to avoid
    interfering with currently running processes.

    Args:
        max_age_seconds: Maximum age in seconds before an item is deleted.
                        Defaults to 1 hour (3600 seconds).

    Returns:
        Dictionary with cleanup statistics:
        - dirs_deleted: Number of directories deleted
        - files_deleted: Number of files deleted
        - bytes_freed: Approximate bytes freed
        - errors: List of error messages (for debugging, not fatal)
    """
    temp_dir = Path(tempfile.gettempdir())
    cutoff_time = time.time() - max_age_seconds

    stats = {
        "dirs_deleted": 0,
        "files_deleted": 0,
        "bytes_freed": 0,
        "errors": [],
    }

    # Clean up directories
    for pattern in DIR_PATTERNS:
        for item in temp_dir.glob(pattern):
            if not item.is_dir():
                continue
            try:
                mtime = item.stat().st_mtime
                if mtime < cutoff_time:
                    size = _get_dir_size(item)
                    shutil.rmtree(item, ignore_errors=True)
                    if not item.exists():
                        stats["dirs_deleted"] += 1
                        stats["bytes_freed"] += size
                        logger.debug(f"Deleted temp directory: {item}")
            except Exception as e:
                stats["errors"].append(f"Failed to delete {item}: {e}")
                logger.debug(f"Failed to delete {item}: {e}")

    # Clean up files
    for pattern in FILE_PATTERNS:
        for item in temp_dir.glob(pattern):
            if not item.is_file():
                continue
            try:
                mtime = item.stat().st_mtime
                if mtime < cutoff_time:
                    size = item.stat().st_size
                    item.unlink(missing_ok=True)
                    if not item.exists():
                        stats["files_deleted"] += 1
                        stats["bytes_freed"] += size
                        logger.debug(f"Deleted temp file: {item}")
            except Exception as e:
                stats["errors"].append(f"Failed to delete {item}: {e}")
                logger.debug(f"Failed to delete {item}: {e}")

    # Log summary if anything was cleaned
    if stats["dirs_deleted"] > 0 or stats["files_deleted"] > 0:
        mb_freed = stats["bytes_freed"] / (1024 * 1024)
        logger.info(
            f"Temp cleanup: {stats['dirs_deleted']} dirs, "
            f"{stats['files_deleted']} files, {mb_freed:.1f} MB freed"
        )

    return stats


def cleanup_project_screenshots(project_dir: Path, max_age_seconds: int = 300) -> dict:
    """
    Clean up stale Playwright CLI artifacts from the project.

    The Playwright CLI daemon saves screenshots, snapshots, and other artifacts
    to `{project_dir}/.playwright-cli/`. This removes them after they've aged
    out (default 5 minutes).

    Also cleans up legacy screenshot patterns from the project root (from the
    old Playwright MCP server approach).

    Args:
        project_dir: Path to the project directory.
        max_age_seconds: Maximum age in seconds before an artifact is deleted.
                        Defaults to 5 minutes (300 seconds).

    Returns:
        Dictionary with cleanup statistics (files_deleted, bytes_freed, errors).
    """
    cutoff_time = time.time() - max_age_seconds
    stats: dict = {"files_deleted": 0, "bytes_freed": 0, "errors": []}

    # Clean up .playwright-cli/ directory (new CLI approach)
    playwright_cli_dir = project_dir / ".playwright-cli"
    if playwright_cli_dir.exists():
        for item in playwright_cli_dir.iterdir():
            if not item.is_file():
                continue
            try:
                mtime = item.stat().st_mtime
                if mtime < cutoff_time:
                    size = item.stat().st_size
                    item.unlink(missing_ok=True)
                    if not item.exists():
                        stats["files_deleted"] += 1
                        stats["bytes_freed"] += size
                        logger.debug(f"Deleted playwright-cli artifact: {item}")
            except Exception as e:
                stats["errors"].append(f"Failed to delete {item}: {e}")
                logger.debug(f"Failed to delete artifact {item}: {e}")

    # Legacy cleanup: root-level screenshot patterns (from old MCP server approach)
    legacy_patterns = [
        "feature*-*.png",
        "screenshot-*.png",
        "step-*.png",
    ]

    for pattern in legacy_patterns:
        for item in project_dir.glob(pattern):
            if not item.is_file():
                continue
            try:
                mtime = item.stat().st_mtime
                if mtime < cutoff_time:
                    size = item.stat().st_size
                    item.unlink(missing_ok=True)
                    if not item.exists():
                        stats["files_deleted"] += 1
                        stats["bytes_freed"] += size
                        logger.debug(f"Deleted legacy screenshot: {item}")
            except Exception as e:
                stats["errors"].append(f"Failed to delete {item}: {e}")
                logger.debug(f"Failed to delete screenshot {item}: {e}")

    if stats["files_deleted"] > 0:
        mb_freed = stats["bytes_freed"] / (1024 * 1024)
        logger.info(f"Artifact cleanup: {stats['files_deleted']} files, {mb_freed:.1f} MB freed")

    return stats


def _get_dir_size(path: Path) -> int:
    """Get total size of a directory in bytes."""
    total = 0
    try:
        for item in path.rglob("*"):
            if item.is_file():
                try:
                    total += item.stat().st_size
                except (OSError, PermissionError):
                    pass
    except (OSError, PermissionError):
        pass
    return total


if __name__ == "__main__":
    # Allow running directly for testing/manual cleanup
    logging.basicConfig(level=logging.DEBUG)
    print("Running temp cleanup...")
    stats = cleanup_stale_temp()
    mb_freed = stats["bytes_freed"] / (1024 * 1024)
    print(f"Cleanup complete: {stats['dirs_deleted']} dirs, {stats['files_deleted']} files, {mb_freed:.1f} MB freed")
    if stats["errors"]:
        print(f"Errors (non-fatal): {len(stats['errors'])}")