fix(automation): purge expired local-mode run workspaces

openhands/automation/app.py

@@ -37,6 +37,7 @@
 from openhands.automation.utils.version import get_sdk_version, get_server_version_info
 from openhands.automation.watchdog import watchdog_loop
 from openhands.automation.webhook_router import router as webhook_router
+from openhands.automation.workspace_cleaner import purger_loop
 
 
 logger = logging.getLogger("automation.app")
@@ -162,18 +163,40 @@ async def lifespan(app: FastAPI):
     app.state.watchdog_task = watchdog_task
     logger.info("Background watchdog started")
 
+    # Purger: removes old workspace directories in local mode only
+    purger_task: asyncio.Task | None = None
+    if settings.is_local_mode:
+        purger_task = asyncio.create_task(
+            purger_loop(
+                app.state.session_factory,
+                workspace_base=os.path.expanduser(
+                    settings.workspace_base or "/workspace"
+                ),
+                retention_seconds=settings.workspace_retention_seconds,
+                interval_seconds=settings.purger_interval_seconds,
+                batch_size=settings.purger_batch_size,
+                shutdown_event=shutdown_event,
+            )
+        )
+        app.state.purger_task = purger_task
+        logger.info("Background workspace purger started")
+
     yield
 
     # Shutdown
     logger.info("Shutting down background tasks...")
     shutdown_event.set()
 
     # Wait for all tasks to exit gracefully
-    for task_name, task in [
+    shutdown_tasks: list[tuple[str, asyncio.Task | None]] = [
         ("scheduler", scheduler_task),
         ("dispatcher", dispatcher_task),
         ("watchdog", watchdog_task),
-    ]:
+        ("purger", purger_task),
+    ]
+    for task_name, task in shutdown_tasks:
+        if task is None:
+            continue
         try:
             await asyncio.wait_for(task, timeout=5.0)
         except TimeoutError:

openhands/automation/config.py

@@ -48,7 +48,7 @@
 from typing import Literal
 from urllib.parse import urlparse
 
-from pydantic import model_validator
+from pydantic import Field, model_validator
 from pydantic_settings import BaseSettings
 
 
@@ -332,6 +332,13 @@ class ServiceSettings(BaseSettings):
         AUTOMATION_DISPATCHER_INTERVAL_SECONDS: Dispatcher poll interval (default: 10)
         AUTOMATION_DISPATCHER_BATCH_SIZE: Dispatcher batch size (default: 10)
         AUTOMATION_WATCHDOG_INTERVAL_SECONDS: Watchdog poll interval (default: 60)
+        AUTOMATION_PURGER_INTERVAL_SECONDS: Workspace purger interval
+            (local mode only, default: 3600 — 1 hour)
+        AUTOMATION_PURGER_BATCH_SIZE: Max workspaces to purge per cycle (default: 50)
+
+        # Workspace retention (local mode only)
+        AUTOMATION_WORKSPACE_RETENTION_SECONDS: Delete workspace directories
+            for terminal runs older than this (default: 604800 — 7 days).
 
         # API pagination
         AUTOMATION_API_DEFAULT_PAGE_SIZE: Default page size (default: 50)
@@ -416,6 +423,11 @@ class ServiceSettings(BaseSettings):
     dispatcher_interval_seconds: int = 10
     dispatcher_batch_size: int = 10
     watchdog_interval_seconds: int = 60
+    purger_interval_seconds: int = Field(default=3600, gt=0)  # 1 hour
+    purger_batch_size: int = Field(default=50, gt=0)
+
+    # Workspace retention for local mode
+    workspace_retention_seconds: int = Field(default=604800, ge=0)  # 7 days
 
     # API pagination
     api_default_page_size: int = 50

openhands/automation/workspace_cleaner.py

@@ -0,0 +1,294 @@
+"""Periodic workspace purging for local-mode terminal runs.
+
+Purging is independent from database-row retention. It only removes
+filesystem workspace directories; database rows are managed separately.
+"""
+
+import asyncio
+import logging
+import os
+import shutil
+from dataclasses import dataclass
+from datetime import timedelta
+from enum import Enum
+from pathlib import Path
+from uuid import UUID
+
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
+
+from openhands.automation.models import AutomationRun, AutomationRunStatus
+from openhands.automation.utils import utcnow
+
+
+logger = logging.getLogger("automation.workspace_cleaner")
+
+TERMINAL_STATES = frozenset(
+    {
+        AutomationRunStatus.COMPLETED,
+        AutomationRunStatus.FAILED,
+        AutomationRunStatus.CANCELLED,
+        AutomationRunStatus.SKIPPED,
+    }
+)
+
+
+@dataclass
+class PurgeResult:
+    """Result of a workspace purge run."""
+
+    candidates_found: int = 0
+    deleted: int = 0
+    missing: int = 0
+    refused: int = 0
+    errors: int = 0
+    bytes_freed: int = 0
+
+
+class DeleteOutcome(Enum):
+    """Outcome of one bounded workspace deletion attempt."""
+
+    DELETED = "deleted"
+    MISSING = "missing"
+    REFUSED = "refused"
+    ERROR = "error"
+
+
+@dataclass(frozen=True)
+class WorkspaceDeleteResult:
+    outcome: DeleteOutcome
+    bytes_freed: int = 0
+
+
+def _workspace_root(workspace_base: str | Path) -> Path:
+    """Return the normalized root that owns all automation run directories.
+
+    Resolve the configured base, but deliberately do not resolve the
+    ``automation-runs`` child. The deletion guard must still be able to detect
+    if that child has been replaced with a symlink or Windows junction.
+    """
+    return Path(workspace_base).expanduser().resolve(strict=False) / "automation-runs"
+
+
+def _workspace_path(workspace_base: str | Path, run_id: UUID) -> Path:
+    """Build a run path from a typed database UUID, never from raw path input."""
+    return _workspace_root(workspace_base) / str(run_id)
+
+
+def _dir_size(path: Path) -> int:
+    total = 0
+    try:
+        for dirpath, _dirnames, filenames in os.walk(path):
+            for filename in filenames:
+                filepath = Path(dirpath) / filename
+                try:
+                    total += os.path.getsize(filepath)
+                except OSError:
+                    pass
+    except OSError:
+        pass
+    return total
+
+
+def _is_link_or_junction(path: Path) -> bool:
+    """Return whether path is a symlink or Windows directory junction."""
+    is_junction = getattr(path, "is_junction", None)
+    return path.is_symlink() or (is_junction is not None and is_junction())
+
+
+def _delete_workspace(
+    workspace_base: str | Path,
+    run_id: UUID,
+) -> WorkspaceDeleteResult:
+    """Delete one verified run directory without following root reparse points."""
+    runs_root = _workspace_root(workspace_base)
+    workspace_path = _workspace_path(workspace_base, run_id)
+
+    if _is_link_or_junction(runs_root):
+        logger.warning("Refusing linked workspace root: %s", runs_root)
+        return WorkspaceDeleteResult(DeleteOutcome.REFUSED)
+
+    if _is_link_or_junction(workspace_path):
+        logger.warning("Refusing linked workspace path: %s", workspace_path)
+        return WorkspaceDeleteResult(DeleteOutcome.REFUSED)
+
+    if not workspace_path.exists():
+        return WorkspaceDeleteResult(DeleteOutcome.MISSING)
+
+    try:
+        resolved_root = runs_root.resolve(strict=True)
+        resolved_path = workspace_path.resolve(strict=True)
+    except FileNotFoundError:
+        return WorkspaceDeleteResult(DeleteOutcome.MISSING)
+    except OSError as exc:
+        logger.warning("Failed to resolve workspace %s: %s", workspace_path, exc)
+        return WorkspaceDeleteResult(DeleteOutcome.ERROR)
+
+    if resolved_path.parent != resolved_root or not resolved_path.is_dir():
+        logger.warning("Refusing workspace outside expected root: %s", workspace_path)
+        return WorkspaceDeleteResult(DeleteOutcome.REFUSED)
+
+    size = _dir_size(workspace_path)
+    try:
+        shutil.rmtree(workspace_path)
+        return WorkspaceDeleteResult(DeleteOutcome.DELETED, size)
+    except FileNotFoundError:
+        return WorkspaceDeleteResult(DeleteOutcome.MISSING)
+    except OSError as exc:
+        logger.warning("Failed to delete workspace %s: %s", workspace_path, exc)
+        return WorkspaceDeleteResult(DeleteOutcome.ERROR)
+
+
+async def purge_terminal_workspaces(
+    session_factory: async_sessionmaker[AsyncSession],
+    workspace_base: str,
+    retention_seconds: int,
+    batch_size: int = 50,
+) -> PurgeResult:
+    """Purge workspace directories for terminal runs past the retention period.
+
+    Only removes filesystem directories; database rows are not touched.
+
+    Never removes workspaces for pending or running runs. Only runs in a
+    terminal state (COMPLETED, FAILED, CANCELLED, SKIPPED) with a
+    ``completed_at`` older than the retention cutoff are eligible.
+
+    Args:
+        session_factory: Factory for async database sessions.
+        workspace_base: Expanded base directory for workspaces.
+        retention_seconds: Minimum age in seconds before a workspace is purged.
+        batch_size: Maximum number of existing workspaces to delete or attempt per
+            call. Missing directories do not consume the limit, preventing old
+            retained database rows from starving later cleanup candidates.
+
+    Returns:
+        PurgeResult with counts of candidates, deletions, errors, and bytes freed.
+    """
+    if retention_seconds < 0:
+        raise ValueError("retention_seconds must be non-negative")
+    if batch_size <= 0:
+        raise ValueError("batch_size must be positive")
+
+    cutoff = utcnow() - timedelta(seconds=retention_seconds)
+    result = PurgeResult()
+
+    async with session_factory() as session:
+        stmt = (
+            select(AutomationRun.id)
+            .where(
+                AutomationRun.status.in_(TERMINAL_STATES),
+                AutomationRun.completed_at.isnot(None),
+                AutomationRun.completed_at < cutoff,
+            )
+            .order_by(AutomationRun.completed_at.asc(), AutomationRun.id.asc())
+            .execution_options(yield_per=batch_size)
+        )
+        candidate_ids = await session.stream_scalars(stmt)
+
+        async for run_id in candidate_ids:
+            result.candidates_found += 1
+            delete_result = await asyncio.to_thread(
+                _delete_workspace, workspace_base, run_id
+            )
+
+            if delete_result.outcome is DeleteOutcome.MISSING:
+                # Missing workspaces are expected after manual cleanup. Continue
+                # scanning so old missing rows cannot starve later real directories.
+                result.missing += 1
+                continue
+            if delete_result.outcome is DeleteOutcome.REFUSED:
+                result.refused += 1
+            elif delete_result.outcome is DeleteOutcome.ERROR:
+                result.errors += 1
+            else:
+                result.deleted += 1
+                result.bytes_freed += delete_result.bytes_freed
+                logger.debug(
+                    "Purged workspace for run %s (%d bytes freed)",
+                    run_id,
+                    delete_result.bytes_freed,
+                )
+
+            attempted_existing = result.deleted + result.refused + result.errors
+            if attempted_existing >= batch_size:
+                break
+
+    logger.info(
+        "Purge complete: %d deleted, %d missing, %d refused, %d errors, "
+        "%d bytes freed (%d candidates scanned)",
+        result.deleted,
+        result.missing,
+        result.refused,
+        result.errors,
+        result.bytes_freed,
+        result.candidates_found,
+    )
+    return result
+
+
+async def purger_loop(
+    session_factory: async_sessionmaker[AsyncSession],
+    workspace_base: str,
+    retention_seconds: int,
+    interval_seconds: int,
+    batch_size: int = 50,
+    shutdown_event: asyncio.Event | None = None,
+) -> None:
+    """Periodic loop that purges old terminal-run workspace directories.
+
+    Only runs in local mode. The caller should guard against running this
+    in cloud mode.
+
+    Args:
+        session_factory: Factory for async database sessions.
+        workspace_base: Expanded base directory for workspaces.
+        retention_seconds: Minimum age before a workspace is purged.
+        interval_seconds: Seconds between purge cycles.
+        batch_size: Maximum workspaces purged per cycle.
+        shutdown_event: Optional event to signal graceful shutdown.
+    """
+    if interval_seconds <= 0:
+        raise ValueError("interval_seconds must be positive")
+
+    logger.info(
+        "Workspace purger started: retention=%ds interval=%ds batch=%d",
+        retention_seconds,
+        interval_seconds,
+        batch_size,
+    )
+
+    while True:
+        if shutdown_event is not None and shutdown_event.is_set():
+            logger.info("Workspace purger received shutdown signal, exiting")
+            break
+
+        try:
+            result = await purge_terminal_workspaces(
+                session_factory=session_factory,
+                workspace_base=workspace_base,
+                retention_seconds=retention_seconds,
+                batch_size=batch_size,
+            )
+            if result.deleted > 0 or result.errors > 0 or result.refused > 0:
+                logger.info(
+                    "Purge cycle: %d deleted (%d bytes), %d missing, "
+                    "%d refused, %d errors, %d candidates",
+                    result.deleted,
+                    result.bytes_freed,
+                    result.missing,
+                    result.refused,
+                    result.errors,
+                    result.candidates_found,
+                )
+        except Exception:
+            logger.exception("Error in workspace purge scan")
+
+        if shutdown_event is not None:
+            try:
+                await asyncio.wait_for(shutdown_event.wait(), timeout=interval_seconds)
+                logger.info("Workspace purger received shutdown signal, exiting")
+                break
+            except TimeoutError:
+                pass
+        else:
+            await asyncio.sleep(interval_seconds)