From 8f9223d48562051646c62e6298545ee13bdddcde Mon Sep 17 00:00:00 2001 From: Trung Minh Do Date: Wed, 29 Jul 2026 02:31:33 +0200 Subject: [PATCH 1/4] feat: purge terminal local-mode automation workspaces after configurable retention period Add a periodic workspace purger for local mode that safely removes workspace directories for runs in terminal states (COMPLETED, FAILED, CANCELLED, SKIPPED) after a configurable retention period. - New openhands/automation/workspace_cleaner.py: purge_terminal_workspaces() queries terminal runs with completed_at < now - retention, deletes workspace dirs in bounded batches, logs bytes freed and errors. Never touches PENDING/RUNNING runs. Tolerates missing directories. - New config: AUTOMATION_WORKSPACE_RETENTION_SECONDS (default 7 days), AUTOMATION_PURGER_INTERVAL_SECONDS (default 1 hour), AUTOMATION_PURGER_BATCH_SIZE (default 50). - app.py: purger_loop background task (local mode only), initial startup purge, graceful shutdown. - tests/test_workspace_cleaner.py: 18 tests covering active-run protection, retention cutoff, batch size, missing dirs, all terminal states, empty DB, and workspace_base expansion. Fixes #271 --- openhands/automation/app.py | 70 ++++- openhands/automation/config.py | 12 + openhands/automation/workspace_cleaner.py | 204 +++++++++++++ tests/test_workspace_cleaner.py | 338 ++++++++++++++++++++++ 4 files changed, 622 insertions(+), 2 deletions(-) create mode 100644 openhands/automation/workspace_cleaner.py create mode 100644 tests/test_workspace_cleaner.py diff --git a/openhands/automation/app.py b/openhands/automation/app.py index ac6b8d4..5c04bf7 100644 --- a/openhands/automation/app.py +++ b/openhands/automation/app.py @@ -6,6 +6,7 @@ from contextlib import asynccontextmanager from importlib.metadata import PackageNotFoundError from pathlib import Path +from typing import TYPE_CHECKING from fastapi import FastAPI from fastapi.responses import JSONResponse @@ -37,11 +38,45 @@ 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, + purge_terminal_workspaces, +) logger = logging.getLogger("automation.app") +if TYPE_CHECKING: + from openhands.automation.config import ServiceSettings + + +async def _startup_purge( + session_factory: Any, + settings: "ServiceSettings", + shutdown_event: asyncio.Event, +) -> None: + """Run an initial workspace purge at startup without blocking startup.""" + try: + await asyncio.sleep(2) + if shutdown_event.is_set(): + return + result = await purge_terminal_workspaces( + session_factory=session_factory, + workspace_base=os.path.expanduser(settings.workspace_base or "/workspace"), + retention_seconds=settings.workspace_retention_seconds, + batch_size=settings.purger_batch_size, + ) + if result.deleted > 0: + logger.info( + "Startup purge: %d workspaces deleted (%d bytes freed)", + result.deleted, + result.bytes_freed, + ) + except Exception: + logger.exception("Startup workspace purge failed") + + @asynccontextmanager async def lifespan(app: FastAPI): """Application startup/shutdown lifecycle.""" @@ -162,6 +197,33 @@ 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") + + # Run an initial purge at startup without blocking + asyncio.create_task( + _startup_purge( + app.state.session_factory, + settings, + shutdown_event, + ) + ) + yield # Shutdown @@ -169,11 +231,15 @@ async def lifespan(app: FastAPI): 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: diff --git a/openhands/automation/config.py b/openhands/automation/config.py index ecf770b..9f22ef3 100644 --- a/openhands/automation/config.py +++ b/openhands/automation/config.py @@ -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 = 3600 # 1 hour + purger_batch_size: int = 50 + + # Workspace retention for local mode + workspace_retention_seconds: int = 604800 # 7 days # API pagination api_default_page_size: int = 50 diff --git a/openhands/automation/workspace_cleaner.py b/openhands/automation/workspace_cleaner.py new file mode 100644 index 0000000..d08ffd1 --- /dev/null +++ b/openhands/automation/workspace_cleaner.py @@ -0,0 +1,204 @@ +"""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 datetime, timedelta + +from sqlalchemy import select +from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker + +from openhands.automation.models import AutomationRun, AutomationRunStatus + + +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 + errors: int = 0 + bytes_freed: int = 0 + + +def _workspace_path(workspace_base: str, run_id: str) -> str: + return os.path.join(workspace_base, "automation-runs", run_id) + + +def _dir_size(path: str) -> int: + total = 0 + try: + for dirpath, _dirnames, filenames in os.walk(path): + for filename in filenames: + filepath = os.path.join(dirpath, filename) + try: + total += os.path.getsize(filepath) + except OSError: + pass + except OSError: + pass + return total + + +def _delete_workspace(workspace_path: str) -> int | None: + """Delete a workspace directory. Returns bytes freed or None on error.""" + try: + size = _dir_size(workspace_path) + except OSError: + size = 0 + + try: + shutil.rmtree(workspace_path) + return size + except FileNotFoundError: + return 0 + except OSError as e: + logger.warning("Failed to delete workspace %s: %s", workspace_path, e) + return None + + +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 workspaces to purge per call. + + Returns: + PurgeResult with counts of candidates, deletions, errors, and bytes freed. + """ + cutoff = datetime.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()) + .limit(batch_size) + ) + rows = await session.execute(stmt) + candidate_ids = [str(row[0]) for row in rows.fetchall()] + + result.candidates_found = len(candidate_ids) + if not candidate_ids: + return result + + logger.info("Found %d candidate workspaces for purge", result.candidates_found) + + for run_id in candidate_ids: + workspace_path = _workspace_path(workspace_base, run_id) + bytes_freed = _delete_workspace(workspace_path) + if bytes_freed is not None: + result.deleted += 1 + result.bytes_freed += bytes_freed + logger.debug( + "Purged workspace %s (%d bytes freed)", workspace_path, bytes_freed + ) + else: + result.errors += 1 + + logger.info( + "Purge complete: %d deleted, %d errors, %d bytes freed", + result.deleted, + result.errors, + result.bytes_freed, + ) + 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. + """ + 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: + logger.info( + "Purge cycle: %d deleted (%d bytes), %d errors, %d candidates", + result.deleted, + result.bytes_freed, + 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) diff --git a/tests/test_workspace_cleaner.py b/tests/test_workspace_cleaner.py new file mode 100644 index 0000000..80ff024 --- /dev/null +++ b/tests/test_workspace_cleaner.py @@ -0,0 +1,338 @@ +"""Tests for workspace purging of local-mode terminal runs.""" + +import os +import uuid +from datetime import datetime, timedelta + +import pytest +from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker + +from openhands.automation.models import Automation, AutomationRun, AutomationRunStatus +from openhands.automation.workspace_cleaner import ( + PurgeResult, + _delete_workspace, + _dir_size, + _workspace_path, + purge_terminal_workspaces, +) + + +async def _create_automation( + session_factory: async_sessionmaker[AsyncSession], + auto_id: uuid.UUID, +) -> None: + async with session_factory() as session: + auto = Automation( + id=auto_id, + user_id=uuid.uuid4(), + org_id=uuid.uuid4(), + name="test-auto", + trigger={"type": "manual"}, + tarball_path="s3://test/tarball.tar.gz", + entrypoint="echo hello", + ) + session.add(auto) + await session.commit() + + +def _run_id() -> uuid.UUID: + return uuid.uuid4() + + +async def _create_run( + session_factory: async_sessionmaker[AsyncSession], + run_id: uuid.UUID, + automation_id: uuid.UUID, + status: AutomationRunStatus, + completed_at: datetime | None = None, +) -> None: + async with session_factory() as session: + run = AutomationRun( + id=run_id, + automation_id=automation_id, + status=status, + completed_at=completed_at, + ) + session.add(run) + await session.commit() + + +def _make_workspace(base: str, run_id: str, content: str = "data") -> str: + path = _workspace_path(base, run_id) + os.makedirs(path, exist_ok=True) + file_path = os.path.join(path, "output.txt") + with open(file_path, "w") as f: + f.write(content) + return path + + +class TestWorkspacePath: + def test_joins_base_and_run_id(self): + path = _workspace_path("/ws", "abc-123") + assert path == os.path.join("/ws", "automation-runs", "abc-123") + + +class TestDirSize: + def test_empty_dir(self, workspace_base): + path = _workspace_path(workspace_base, "empty-run") + os.makedirs(path, exist_ok=True) + assert _dir_size(path) == 0 + + def test_non_empty_dir(self, workspace_base): + path = _make_workspace(workspace_base, "data-run", "hello world") + size = _dir_size(path) + assert size == len("hello world") + + def test_missing_dir(self, workspace_base): + path = _workspace_path(workspace_base, "missing-run") + assert _dir_size(path) == 0 + + +class TestDeleteWorkspace: + def test_deletes_existing(self, workspace_base): + path = _make_workspace(workspace_base, "del-run") + assert os.path.isdir(path) + bytes_freed = _delete_workspace(path) + assert bytes_freed is not None + assert bytes_freed > 0 + assert not os.path.exists(path) + + def test_missing_is_idempotent(self, workspace_base): + path = _workspace_path(workspace_base, "already-gone") + bytes_freed = _delete_workspace(path) + assert bytes_freed == 0 + + def test_empty_dir(self, workspace_base): + path = _workspace_path(workspace_base, "empty") + os.makedirs(path, exist_ok=True) + bytes_freed = _delete_workspace(path) + assert bytes_freed == 0 + assert not os.path.exists(path) + + +class TestPurgeTerminalWorkspaces: + async def test_never_purges_active_runs(self, db_session_factory, workspace_base): + """PENDING and RUNNING runs must never be purged.""" + auto_id = uuid.uuid4() + await _create_automation(db_session_factory, auto_id) + + pending_id = _run_id() + running_id = _run_id() + await _create_run( + db_session_factory, pending_id, auto_id, AutomationRunStatus.PENDING + ) + await _create_run( + db_session_factory, running_id, auto_id, AutomationRunStatus.RUNNING + ) + + result = await purge_terminal_workspaces( + db_session_factory, workspace_base, retention_seconds=0, batch_size=10 + ) + + assert result.candidates_found == 0 + assert result.deleted == 0 + + async def test_respects_retention_cutoff(self, db_session_factory, workspace_base): + """Recent terminal runs must not be purged.""" + auto_id = uuid.uuid4() + await _create_automation(db_session_factory, auto_id) + + run_id = _run_id() + await _create_run( + db_session_factory, + run_id, + auto_id, + AutomationRunStatus.COMPLETED, + completed_at=datetime.utcnow() - timedelta(seconds=60), + ) + _make_workspace(workspace_base, str(run_id)) + + result = await purge_terminal_workspaces( + db_session_factory, workspace_base, retention_seconds=3600, batch_size=10 + ) + + assert result.candidates_found == 0 + assert result.deleted == 0 + + async def test_purges_old_terminal_workspace( + self, db_session_factory, workspace_base + ): + """Old terminal run workspaces must be purged.""" + auto_id = uuid.uuid4() + await _create_automation(db_session_factory, auto_id) + + run_id = _run_id() + await _create_run( + db_session_factory, + run_id, + auto_id, + AutomationRunStatus.FAILED, + completed_at=datetime.utcnow() - timedelta(days=30), + ) + path = _make_workspace(workspace_base, str(run_id), "some data here") + assert os.path.isdir(path) + + result = await purge_terminal_workspaces( + db_session_factory, workspace_base, retention_seconds=3600, batch_size=10 + ) + + assert result.candidates_found == 1 + assert result.deleted == 1 + assert result.errors == 0 + assert result.bytes_freed == len("some data here") + assert not os.path.exists(path) + + async def test_all_terminal_states(self, db_session_factory, workspace_base): + """All terminal states are purged.""" + auto_id = uuid.uuid4() + await _create_automation(db_session_factory, auto_id) + + old_time = datetime.utcnow() - timedelta(days=30) + for status in ( + AutomationRunStatus.COMPLETED, + AutomationRunStatus.FAILED, + AutomationRunStatus.CANCELLED, + AutomationRunStatus.SKIPPED, + ): + run_id = _run_id() + await _create_run( + db_session_factory, run_id, auto_id, status, completed_at=old_time + ) + _make_workspace(workspace_base, str(run_id)) + + result = await purge_terminal_workspaces( + db_session_factory, workspace_base, retention_seconds=3600, batch_size=10 + ) + + assert result.candidates_found == 4 + assert result.deleted == 4 + assert result.errors == 0 + + async def test_tolerates_missing_workspace( + self, db_session_factory, workspace_base + ): + """Runs without workspace directories must not cause errors.""" + auto_id = uuid.uuid4() + await _create_automation(db_session_factory, auto_id) + + run_id = _run_id() + await _create_run( + db_session_factory, + run_id, + auto_id, + AutomationRunStatus.COMPLETED, + completed_at=datetime.utcnow() - timedelta(days=30), + ) + + result = await purge_terminal_workspaces( + db_session_factory, workspace_base, retention_seconds=3600, batch_size=10 + ) + + assert result.candidates_found == 1 + assert result.deleted == 1 + assert result.errors == 0 + + async def test_respects_batch_size(self, db_session_factory, workspace_base): + """Only batch_size workspaces are purged per call.""" + auto_id = uuid.uuid4() + await _create_automation(db_session_factory, auto_id) + + old_time = datetime.utcnow() - timedelta(days=30) + for _ in range(5): + run_id = _run_id() + await _create_run( + db_session_factory, + run_id, + auto_id, + AutomationRunStatus.COMPLETED, + completed_at=old_time, + ) + _make_workspace(workspace_base, str(run_id)) + + result = await purge_terminal_workspaces( + db_session_factory, workspace_base, retention_seconds=3600, batch_size=3 + ) + + assert result.candidates_found == 3 + assert result.deleted == 3 + + async def test_empty_database(self, db_session_factory, workspace_base): + """Purger handles an empty database gracefully.""" + result = await purge_terminal_workspaces( + db_session_factory, workspace_base, retention_seconds=3600, batch_size=10 + ) + + assert result.candidates_found == 0 + assert result.deleted == 0 + assert result.errors == 0 + + async def test_skips_runs_without_completed_at( + self, db_session_factory, workspace_base + ): + """Terminal runs without completed_at are skipped.""" + auto_id = uuid.uuid4() + await _create_automation(db_session_factory, auto_id) + + run_id = _run_id() + await _create_run( + db_session_factory, + run_id, + auto_id, + AutomationRunStatus.CANCELLED, + completed_at=None, + ) + _make_workspace(workspace_base, str(run_id)) + + result = await purge_terminal_workspaces( + db_session_factory, workspace_base, retention_seconds=0, batch_size=10 + ) + + assert result.candidates_found == 0 + assert result.deleted == 0 + + async def test_workspace_base_expansion(self, db_session_factory): + """`~` in workspace_base is expanded.""" + auto_id = uuid.uuid4() + await _create_automation(db_session_factory, auto_id) + + old_time = datetime.utcnow() - timedelta(days=30) + run_id = _run_id() + await _create_run( + db_session_factory, + run_id, + auto_id, + AutomationRunStatus.COMPLETED, + completed_at=old_time, + ) + + result = await purge_terminal_workspaces( + db_session_factory, + os.path.expanduser("~"), + retention_seconds=3600, + batch_size=10, + ) + + assert result.candidates_found == 1 + assert result.deleted == 1 + assert result.errors == 0 + + +class TestPurgeResult: + def test_default_values(self): + result = PurgeResult() + assert result.candidates_found == 0 + assert result.deleted == 0 + assert result.errors == 0 + assert result.bytes_freed == 0 + + def test_partial_success(self): + result = PurgeResult( + candidates_found=10, + deleted=8, + errors=2, + bytes_freed=1024, + ) + assert result.candidates_found == 10 + assert result.deleted == 8 + assert result.errors == 2 + assert result.bytes_freed == 1024 From b07b6d97ac7885de7c12ee28e53ce618b09c0e6a Mon Sep 17 00:00:00 2001 From: Trung Minh Do Date: Thu, 30 Jul 2026 01:45:03 +0200 Subject: [PATCH 2/4] fix: harden local workspace retention cleanup --- openhands/automation/app.py | 45 +-- openhands/automation/config.py | 8 +- openhands/automation/workspace_cleaner.py | 162 +++++++--- tests/test_config.py | 21 ++ tests/test_workspace_cleaner.py | 355 +++++++++++++++++++--- 5 files changed, 456 insertions(+), 135 deletions(-) diff --git a/openhands/automation/app.py b/openhands/automation/app.py index 5c04bf7..1463ad2 100644 --- a/openhands/automation/app.py +++ b/openhands/automation/app.py @@ -6,7 +6,6 @@ from contextlib import asynccontextmanager from importlib.metadata import PackageNotFoundError from pathlib import Path -from typing import TYPE_CHECKING from fastapi import FastAPI from fastapi.responses import JSONResponse @@ -38,45 +37,12 @@ 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, - purge_terminal_workspaces, -) +from openhands.automation.workspace_cleaner import purger_loop logger = logging.getLogger("automation.app") -if TYPE_CHECKING: - from openhands.automation.config import ServiceSettings - - -async def _startup_purge( - session_factory: Any, - settings: "ServiceSettings", - shutdown_event: asyncio.Event, -) -> None: - """Run an initial workspace purge at startup without blocking startup.""" - try: - await asyncio.sleep(2) - if shutdown_event.is_set(): - return - result = await purge_terminal_workspaces( - session_factory=session_factory, - workspace_base=os.path.expanduser(settings.workspace_base or "/workspace"), - retention_seconds=settings.workspace_retention_seconds, - batch_size=settings.purger_batch_size, - ) - if result.deleted > 0: - logger.info( - "Startup purge: %d workspaces deleted (%d bytes freed)", - result.deleted, - result.bytes_freed, - ) - except Exception: - logger.exception("Startup workspace purge failed") - - @asynccontextmanager async def lifespan(app: FastAPI): """Application startup/shutdown lifecycle.""" @@ -215,15 +181,6 @@ async def lifespan(app: FastAPI): app.state.purger_task = purger_task logger.info("Background workspace purger started") - # Run an initial purge at startup without blocking - asyncio.create_task( - _startup_purge( - app.state.session_factory, - settings, - shutdown_event, - ) - ) - yield # Shutdown diff --git a/openhands/automation/config.py b/openhands/automation/config.py index 9f22ef3..aad2f64 100644 --- a/openhands/automation/config.py +++ b/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 @@ -423,11 +423,11 @@ class ServiceSettings(BaseSettings): dispatcher_interval_seconds: int = 10 dispatcher_batch_size: int = 10 watchdog_interval_seconds: int = 60 - purger_interval_seconds: int = 3600 # 1 hour - purger_batch_size: int = 50 + 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 = 604800 # 7 days + workspace_retention_seconds: int = Field(default=604800, ge=0) # 7 days # API pagination api_default_page_size: int = 50 diff --git a/openhands/automation/workspace_cleaner.py b/openhands/automation/workspace_cleaner.py index d08ffd1..08ee90f 100644 --- a/openhands/automation/workspace_cleaner.py +++ b/openhands/automation/workspace_cleaner.py @@ -9,12 +9,16 @@ import os import shutil from dataclasses import dataclass -from datetime import datetime, timedelta +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") @@ -35,20 +39,43 @@ class PurgeResult: candidates_found: int = 0 deleted: int = 0 + missing: int = 0 + refused: int = 0 errors: int = 0 bytes_freed: int = 0 -def _workspace_path(workspace_base: str, run_id: str) -> str: - return os.path.join(workspace_base, "automation-runs", run_id) +class DeleteOutcome(Enum): + """Outcome of one bounded workspace deletion attempt.""" + DELETED = "deleted" + MISSING = "missing" + REFUSED = "refused" + ERROR = "error" -def _dir_size(path: str) -> int: + +@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.""" + return (Path(workspace_base).expanduser() / "automation-runs").resolve(strict=False) + + +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 = os.path.join(dirpath, filename) + filepath = Path(dirpath) / filename try: total += os.path.getsize(filepath) except OSError: @@ -58,21 +85,48 @@ def _dir_size(path: str) -> int: return total -def _delete_workspace(workspace_path: str) -> int | None: - """Delete a workspace directory. Returns bytes freed or None on error.""" +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(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: - size = _dir_size(workspace_path) - except OSError: - size = 0 + 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 != runs_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 size + return WorkspaceDeleteResult(DeleteOutcome.DELETED, size) except FileNotFoundError: - return 0 - except OSError as e: - logger.warning("Failed to delete workspace %s: %s", workspace_path, e) - return None + 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( @@ -93,12 +147,19 @@ async def purge_terminal_workspaces( 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 workspaces to purge per call. + 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. """ - cutoff = datetime.utcnow() - timedelta(seconds=retention_seconds) + 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: @@ -109,35 +170,48 @@ async def purge_terminal_workspaces( AutomationRun.completed_at.isnot(None), AutomationRun.completed_at < cutoff, ) - .order_by(AutomationRun.completed_at.asc()) - .limit(batch_size) + .order_by(AutomationRun.completed_at.asc(), AutomationRun.id.asc()) + .execution_options(yield_per=batch_size) ) - rows = await session.execute(stmt) - candidate_ids = [str(row[0]) for row in rows.fetchall()] - - result.candidates_found = len(candidate_ids) - if not candidate_ids: - return result - - logger.info("Found %d candidate workspaces for purge", result.candidates_found) - - for run_id in candidate_ids: - workspace_path = _workspace_path(workspace_base, run_id) - bytes_freed = _delete_workspace(workspace_path) - if bytes_freed is not None: - result.deleted += 1 - result.bytes_freed += bytes_freed - logger.debug( - "Purged workspace %s (%d bytes freed)", workspace_path, bytes_freed + 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 ) - else: - result.errors += 1 + + 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 errors, %d bytes freed", + "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 @@ -163,6 +237,9 @@ async def purger_loop( 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, @@ -182,11 +259,14 @@ async def purger_loop( retention_seconds=retention_seconds, batch_size=batch_size, ) - if result.deleted > 0 or result.errors > 0: + if result.deleted > 0 or result.errors > 0 or result.refused > 0: logger.info( - "Purge cycle: %d deleted (%d bytes), %d errors, %d candidates", + "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, ) diff --git a/tests/test_config.py b/tests/test_config.py index 1bf78a9..9ef5fac 100644 --- a/tests/test_config.py +++ b/tests/test_config.py @@ -2,6 +2,8 @@ import warnings +import pytest + from openhands.automation.config import ( HttpSettings, LogSettings, @@ -306,6 +308,25 @@ def test_local_mode_full_configuration(self): assert settings.workspace_base == "/my/workspace" assert settings.db_url == "sqlite+aiosqlite:////data/automations.db" + def test_workspace_purger_defaults(self): + settings = Settings() + + assert settings.workspace_retention_seconds == 7 * 24 * 60 * 60 + assert settings.purger_interval_seconds == 60 * 60 + assert settings.purger_batch_size == 50 + + @pytest.mark.parametrize( + ("field", "value"), + [ + ("workspace_retention_seconds", -1), + ("purger_interval_seconds", 0), + ("purger_batch_size", 0), + ], + ) + def test_workspace_purger_rejects_invalid_limits(self, field, value): + with pytest.raises(ValueError): + Settings(**{field: value}) + def test_local_mode_from_env(self, monkeypatch): """Local mode settings are loaded from environment variables.""" monkeypatch.setenv("AUTOMATION_AGENT_SERVER_URL", "http://localhost:3000") diff --git a/tests/test_workspace_cleaner.py b/tests/test_workspace_cleaner.py index 80ff024..3de6c79 100644 --- a/tests/test_workspace_cleaner.py +++ b/tests/test_workspace_cleaner.py @@ -1,22 +1,57 @@ """Tests for workspace purging of local-mode terminal runs.""" -import os import uuid +from collections.abc import AsyncGenerator from datetime import datetime, timedelta +from pathlib import Path +from unittest.mock import AsyncMock import pytest -from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker +from sqlalchemy.ext.asyncio import ( + AsyncSession, + async_sessionmaker, + create_async_engine, +) -from openhands.automation.models import Automation, AutomationRun, AutomationRunStatus +import openhands.automation.workspace_cleaner as cleaner +from openhands.automation.models import ( + Automation, + AutomationRun, + AutomationRunStatus, + Base, +) +from openhands.automation.utils import utcnow from openhands.automation.workspace_cleaner import ( + DeleteOutcome, PurgeResult, _delete_workspace, _dir_size, _workspace_path, purge_terminal_workspaces, + purger_loop, ) +@pytest.fixture +async def db_session_factory() -> AsyncGenerator[ + async_sessionmaker[AsyncSession], None +]: + """Provide a network-free SQLite database for cleanup tests.""" + engine = create_async_engine("sqlite+aiosqlite:///:memory:") + async with engine.begin() as connection: + await connection.run_sync(Base.metadata.create_all) + try: + yield async_sessionmaker(engine, class_=AsyncSession, expire_on_commit=False) + finally: + await engine.dispose() + + +@pytest.fixture +def workspace_base(tmp_path) -> str: + """Keep every filesystem deletion inside the current test temp directory.""" + return str(tmp_path / "workspace") + + async def _create_automation( session_factory: async_sessionmaker[AsyncSession], auto_id: uuid.UUID, @@ -57,57 +92,107 @@ async def _create_run( await session.commit() -def _make_workspace(base: str, run_id: str, content: str = "data") -> str: +def _make_workspace(base: str, run_id: uuid.UUID, content: str = "data") -> Path: path = _workspace_path(base, run_id) - os.makedirs(path, exist_ok=True) - file_path = os.path.join(path, "output.txt") - with open(file_path, "w") as f: - f.write(content) + path.mkdir(parents=True, exist_ok=True) + (path / "output.txt").write_text(content, encoding="utf-8") return path class TestWorkspacePath: def test_joins_base_and_run_id(self): - path = _workspace_path("/ws", "abc-123") - assert path == os.path.join("/ws", "automation-runs", "abc-123") + run_id = _run_id() + path = _workspace_path("/ws", run_id) + assert path == Path("/ws/automation-runs").resolve() / str(run_id) + + def test_native_and_forward_slash_bases_normalize_identically(self, tmp_path): + run_id = _run_id() + native_path = _workspace_path(str(tmp_path), run_id) + forward_slash_path = _workspace_path(tmp_path.as_posix(), run_id) + assert native_path == forward_slash_path class TestDirSize: def test_empty_dir(self, workspace_base): - path = _workspace_path(workspace_base, "empty-run") - os.makedirs(path, exist_ok=True) + path = _workspace_path(workspace_base, _run_id()) + path.mkdir(parents=True) assert _dir_size(path) == 0 def test_non_empty_dir(self, workspace_base): - path = _make_workspace(workspace_base, "data-run", "hello world") + path = _make_workspace(workspace_base, _run_id(), "hello world") size = _dir_size(path) assert size == len("hello world") def test_missing_dir(self, workspace_base): - path = _workspace_path(workspace_base, "missing-run") + path = _workspace_path(workspace_base, _run_id()) assert _dir_size(path) == 0 class TestDeleteWorkspace: def test_deletes_existing(self, workspace_base): - path = _make_workspace(workspace_base, "del-run") - assert os.path.isdir(path) - bytes_freed = _delete_workspace(path) - assert bytes_freed is not None - assert bytes_freed > 0 - assert not os.path.exists(path) + run_id = _run_id() + path = _make_workspace(workspace_base, run_id) + delete_result = _delete_workspace(workspace_base, run_id) + assert delete_result.outcome is DeleteOutcome.DELETED + assert delete_result.bytes_freed > 0 + assert not path.exists() def test_missing_is_idempotent(self, workspace_base): - path = _workspace_path(workspace_base, "already-gone") - bytes_freed = _delete_workspace(path) - assert bytes_freed == 0 + delete_result = _delete_workspace(workspace_base, _run_id()) + assert delete_result.outcome is DeleteOutcome.MISSING + assert delete_result.bytes_freed == 0 def test_empty_dir(self, workspace_base): - path = _workspace_path(workspace_base, "empty") - os.makedirs(path, exist_ok=True) - bytes_freed = _delete_workspace(path) - assert bytes_freed == 0 - assert not os.path.exists(path) + run_id = _run_id() + path = _workspace_path(workspace_base, run_id) + path.mkdir(parents=True) + delete_result = _delete_workspace(workspace_base, run_id) + assert delete_result.outcome is DeleteOutcome.DELETED + assert delete_result.bytes_freed == 0 + assert not path.exists() + + def test_refuses_file_instead_of_directory(self, workspace_base): + run_id = _run_id() + path = _workspace_path(workspace_base, run_id) + path.parent.mkdir(parents=True) + path.write_text("not a workspace directory", encoding="utf-8") + + delete_result = _delete_workspace(workspace_base, run_id) + + assert delete_result.outcome is DeleteOutcome.REFUSED + assert path.exists() + + def test_refuses_symlink_escape(self, workspace_base, tmp_path): + run_id = _run_id() + outside = tmp_path / "outside" + outside.mkdir() + marker = outside / "keep.txt" + marker.write_text("keep", encoding="utf-8") + link = _workspace_path(workspace_base, run_id) + link.parent.mkdir(parents=True) + try: + link.symlink_to(outside, target_is_directory=True) + except OSError as exc: + pytest.skip(f"directory symlinks unavailable: {exc}") + + delete_result = _delete_workspace(workspace_base, run_id) + + assert delete_result.outcome is DeleteOutcome.REFUSED + assert marker.exists() + + def test_refuses_detected_junction(self, workspace_base, monkeypatch): + run_id = _run_id() + path = _make_workspace(workspace_base, run_id) + monkeypatch.setattr( + cleaner, + "_is_link_or_junction", + lambda candidate: candidate == path, + ) + + delete_result = _delete_workspace(workspace_base, run_id) + + assert delete_result.outcome is DeleteOutcome.REFUSED + assert path.exists() class TestPurgeTerminalWorkspaces: @@ -118,12 +203,23 @@ async def test_never_purges_active_runs(self, db_session_factory, workspace_base pending_id = _run_id() running_id = _run_id() + old_time = utcnow() - timedelta(days=30) await _create_run( - db_session_factory, pending_id, auto_id, AutomationRunStatus.PENDING + db_session_factory, + pending_id, + auto_id, + AutomationRunStatus.PENDING, + completed_at=old_time, ) await _create_run( - db_session_factory, running_id, auto_id, AutomationRunStatus.RUNNING + db_session_factory, + running_id, + auto_id, + AutomationRunStatus.RUNNING, + completed_at=old_time, ) + pending_path = _make_workspace(workspace_base, pending_id) + running_path = _make_workspace(workspace_base, running_id) result = await purge_terminal_workspaces( db_session_factory, workspace_base, retention_seconds=0, batch_size=10 @@ -131,6 +227,8 @@ async def test_never_purges_active_runs(self, db_session_factory, workspace_base assert result.candidates_found == 0 assert result.deleted == 0 + assert pending_path.exists() + assert running_path.exists() async def test_respects_retention_cutoff(self, db_session_factory, workspace_base): """Recent terminal runs must not be purged.""" @@ -143,9 +241,9 @@ async def test_respects_retention_cutoff(self, db_session_factory, workspace_bas run_id, auto_id, AutomationRunStatus.COMPLETED, - completed_at=datetime.utcnow() - timedelta(seconds=60), + completed_at=utcnow() - timedelta(seconds=60), ) - _make_workspace(workspace_base, str(run_id)) + path = _make_workspace(workspace_base, run_id) result = await purge_terminal_workspaces( db_session_factory, workspace_base, retention_seconds=3600, batch_size=10 @@ -153,6 +251,7 @@ async def test_respects_retention_cutoff(self, db_session_factory, workspace_bas assert result.candidates_found == 0 assert result.deleted == 0 + assert path.exists() async def test_purges_old_terminal_workspace( self, db_session_factory, workspace_base @@ -167,10 +266,10 @@ async def test_purges_old_terminal_workspace( run_id, auto_id, AutomationRunStatus.FAILED, - completed_at=datetime.utcnow() - timedelta(days=30), + completed_at=utcnow() - timedelta(days=30), ) - path = _make_workspace(workspace_base, str(run_id), "some data here") - assert os.path.isdir(path) + path = _make_workspace(workspace_base, run_id, "some data here") + assert path.is_dir() result = await purge_terminal_workspaces( db_session_factory, workspace_base, retention_seconds=3600, batch_size=10 @@ -180,14 +279,14 @@ async def test_purges_old_terminal_workspace( assert result.deleted == 1 assert result.errors == 0 assert result.bytes_freed == len("some data here") - assert not os.path.exists(path) + assert not path.exists() async def test_all_terminal_states(self, db_session_factory, workspace_base): """All terminal states are purged.""" auto_id = uuid.uuid4() await _create_automation(db_session_factory, auto_id) - old_time = datetime.utcnow() - timedelta(days=30) + old_time = utcnow() - timedelta(days=30) for status in ( AutomationRunStatus.COMPLETED, AutomationRunStatus.FAILED, @@ -198,7 +297,7 @@ async def test_all_terminal_states(self, db_session_factory, workspace_base): await _create_run( db_session_factory, run_id, auto_id, status, completed_at=old_time ) - _make_workspace(workspace_base, str(run_id)) + _make_workspace(workspace_base, run_id) result = await purge_terminal_workspaces( db_session_factory, workspace_base, retention_seconds=3600, batch_size=10 @@ -221,7 +320,7 @@ async def test_tolerates_missing_workspace( run_id, auto_id, AutomationRunStatus.COMPLETED, - completed_at=datetime.utcnow() - timedelta(days=30), + completed_at=utcnow() - timedelta(days=30), ) result = await purge_terminal_workspaces( @@ -229,7 +328,8 @@ async def test_tolerates_missing_workspace( ) assert result.candidates_found == 1 - assert result.deleted == 1 + assert result.deleted == 0 + assert result.missing == 1 assert result.errors == 0 async def test_respects_batch_size(self, db_session_factory, workspace_base): @@ -237,7 +337,7 @@ async def test_respects_batch_size(self, db_session_factory, workspace_base): auto_id = uuid.uuid4() await _create_automation(db_session_factory, auto_id) - old_time = datetime.utcnow() - timedelta(days=30) + old_time = utcnow() - timedelta(days=30) for _ in range(5): run_id = _run_id() await _create_run( @@ -247,7 +347,7 @@ async def test_respects_batch_size(self, db_session_factory, workspace_base): AutomationRunStatus.COMPLETED, completed_at=old_time, ) - _make_workspace(workspace_base, str(run_id)) + _make_workspace(workspace_base, run_id) result = await purge_terminal_workspaces( db_session_factory, workspace_base, retention_seconds=3600, batch_size=3 @@ -255,6 +355,8 @@ async def test_respects_batch_size(self, db_session_factory, workspace_base): assert result.candidates_found == 3 assert result.deleted == 3 + remaining = list((Path(workspace_base) / "automation-runs").iterdir()) + assert len(remaining) == 2 async def test_empty_database(self, db_session_factory, workspace_base): """Purger handles an empty database gracefully.""" @@ -281,7 +383,7 @@ async def test_skips_runs_without_completed_at( AutomationRunStatus.CANCELLED, completed_at=None, ) - _make_workspace(workspace_base, str(run_id)) + path = _make_workspace(workspace_base, run_id) result = await purge_terminal_workspaces( db_session_factory, workspace_base, retention_seconds=0, batch_size=10 @@ -289,13 +391,18 @@ async def test_skips_runs_without_completed_at( assert result.candidates_found == 0 assert result.deleted == 0 + assert path.exists() - async def test_workspace_base_expansion(self, db_session_factory): + async def test_workspace_base_expansion( + self, db_session_factory, tmp_path, monkeypatch + ): """`~` in workspace_base is expanded.""" + monkeypatch.setenv("HOME", str(tmp_path)) + monkeypatch.setenv("USERPROFILE", str(tmp_path)) auto_id = uuid.uuid4() await _create_automation(db_session_factory, auto_id) - old_time = datetime.utcnow() - timedelta(days=30) + old_time = utcnow() - timedelta(days=30) run_id = _run_id() await _create_run( db_session_factory, @@ -304,10 +411,11 @@ async def test_workspace_base_expansion(self, db_session_factory): AutomationRunStatus.COMPLETED, completed_at=old_time, ) + path = _make_workspace("~", run_id) result = await purge_terminal_workspaces( db_session_factory, - os.path.expanduser("~"), + "~", retention_seconds=3600, batch_size=10, ) @@ -315,6 +423,155 @@ async def test_workspace_base_expansion(self, db_session_factory): assert result.candidates_found == 1 assert result.deleted == 1 assert result.errors == 0 + assert not path.exists() + + async def test_missing_old_row_does_not_starve_existing_workspace( + self, db_session_factory, workspace_base + ): + auto_id = uuid.uuid4() + await _create_automation(db_session_factory, auto_id) + old_time = utcnow() - timedelta(days=30) + + missing_id = uuid.UUID(int=1) + existing_id = uuid.UUID(int=2) + await _create_run( + db_session_factory, + missing_id, + auto_id, + AutomationRunStatus.COMPLETED, + completed_at=old_time - timedelta(seconds=1), + ) + await _create_run( + db_session_factory, + existing_id, + auto_id, + AutomationRunStatus.COMPLETED, + completed_at=old_time, + ) + existing_path = _make_workspace(workspace_base, existing_id) + + result = await purge_terminal_workspaces( + db_session_factory, workspace_base, retention_seconds=3600, batch_size=1 + ) + + assert result.candidates_found == 2 + assert result.missing == 1 + assert result.deleted == 1 + assert not existing_path.exists() + + async def test_partially_created_workspace_is_deleted( + self, db_session_factory, workspace_base + ): + auto_id = uuid.uuid4() + await _create_automation(db_session_factory, auto_id) + run_id = _run_id() + await _create_run( + db_session_factory, + run_id, + auto_id, + AutomationRunStatus.FAILED, + completed_at=utcnow() - timedelta(days=30), + ) + path = _workspace_path(workspace_base, run_id) + path.mkdir(parents=True) + + result = await purge_terminal_workspaces( + db_session_factory, workspace_base, retention_seconds=3600, batch_size=1 + ) + + assert result.deleted == 1 + assert result.bytes_freed == 0 + assert not path.exists() + + async def test_delete_error_is_retryable( + self, db_session_factory, workspace_base, monkeypatch + ): + auto_id = uuid.uuid4() + await _create_automation(db_session_factory, auto_id) + run_id = _run_id() + await _create_run( + db_session_factory, + run_id, + auto_id, + AutomationRunStatus.COMPLETED, + completed_at=utcnow() - timedelta(days=30), + ) + path = _make_workspace(workspace_base, run_id) + real_rmtree = cleaner.shutil.rmtree + + def fail_rmtree(_path): + raise PermissionError("locked") + + monkeypatch.setattr(cleaner.shutil, "rmtree", fail_rmtree) + + first_result = await purge_terminal_workspaces( + db_session_factory, workspace_base, retention_seconds=3600, batch_size=1 + ) + + assert first_result.errors == 1 + assert first_result.deleted == 0 + assert path.exists() + + monkeypatch.setattr(cleaner.shutil, "rmtree", real_rmtree) + retry_result = await purge_terminal_workspaces( + db_session_factory, workspace_base, retention_seconds=3600, batch_size=1 + ) + + assert retry_result.deleted == 1 + assert retry_result.errors == 0 + assert not path.exists() + + @pytest.mark.parametrize( + ("retention_seconds", "batch_size", "message"), + [(-1, 1, "retention_seconds"), (0, 0, "batch_size")], + ) + async def test_rejects_invalid_limits( + self, + db_session_factory, + workspace_base, + retention_seconds, + batch_size, + message, + ): + with pytest.raises(ValueError, match=message): + await purge_terminal_workspaces( + db_session_factory, + workspace_base, + retention_seconds=retention_seconds, + batch_size=batch_size, + ) + + +class TestPurgerLoop: + async def test_runs_immediately_then_stops_on_shutdown(self, monkeypatch): + shutdown_event = cleaner.asyncio.Event() + + async def purge_once(**_kwargs): + shutdown_event.set() + return PurgeResult() + + mock_purge = AsyncMock(side_effect=purge_once) + monkeypatch.setattr(cleaner, "purge_terminal_workspaces", mock_purge) + + await purger_loop( + session_factory=AsyncMock(), + workspace_base="/workspace", + retention_seconds=60, + interval_seconds=3600, + batch_size=10, + shutdown_event=shutdown_event, + ) + + mock_purge.assert_awaited_once() + + async def test_rejects_non_positive_interval(self): + with pytest.raises(ValueError, match="interval_seconds"): + await purger_loop( + session_factory=AsyncMock(), + workspace_base="/workspace", + retention_seconds=60, + interval_seconds=0, + ) class TestPurgeResult: @@ -322,6 +579,8 @@ def test_default_values(self): result = PurgeResult() assert result.candidates_found == 0 assert result.deleted == 0 + assert result.missing == 0 + assert result.refused == 0 assert result.errors == 0 assert result.bytes_freed == 0 @@ -329,10 +588,14 @@ def test_partial_success(self): result = PurgeResult( candidates_found=10, deleted=8, + missing=1, + refused=1, errors=2, bytes_freed=1024, ) assert result.candidates_found == 10 assert result.deleted == 8 + assert result.missing == 1 + assert result.refused == 1 assert result.errors == 2 assert result.bytes_freed == 1024 From ed201ddbfe7ab71dfe52a2e6820a94a8a894c3d0 Mon Sep 17 00:00:00 2001 From: Trung Minh Do Date: Thu, 30 Jul 2026 02:47:52 +0200 Subject: [PATCH 3/4] fix(automation): harden local workspace retention --- openhands/automation/workspace_cleaner.py | 16 ++++++++-- tests/test_workspace_cleaner.py | 36 +++++++++++++++++++++++ 2 files changed, 49 insertions(+), 3 deletions(-) diff --git a/openhands/automation/workspace_cleaner.py b/openhands/automation/workspace_cleaner.py index 08ee90f..aec46b4 100644 --- a/openhands/automation/workspace_cleaner.py +++ b/openhands/automation/workspace_cleaner.py @@ -61,8 +61,13 @@ class WorkspaceDeleteResult: def _workspace_root(workspace_base: str | Path) -> Path: - """Return the normalized root that owns all automation run directories.""" - return (Path(workspace_base).expanduser() / "automation-runs").resolve(strict=False) + """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: @@ -99,6 +104,10 @@ def _delete_workspace( 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) @@ -107,6 +116,7 @@ def _delete_workspace( 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) @@ -114,7 +124,7 @@ def _delete_workspace( logger.warning("Failed to resolve workspace %s: %s", workspace_path, exc) return WorkspaceDeleteResult(DeleteOutcome.ERROR) - if resolved_path.parent != runs_root or not resolved_path.is_dir(): + 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) diff --git a/tests/test_workspace_cleaner.py b/tests/test_workspace_cleaner.py index 3de6c79..0f0b4d6 100644 --- a/tests/test_workspace_cleaner.py +++ b/tests/test_workspace_cleaner.py @@ -180,6 +180,25 @@ def test_refuses_symlink_escape(self, workspace_base, tmp_path): assert delete_result.outcome is DeleteOutcome.REFUSED assert marker.exists() + def test_refuses_symlinked_workspace_root(self, workspace_base, tmp_path): + run_id = _run_id() + outside = tmp_path / "outside-root" + workspace = outside / str(run_id) + workspace.mkdir(parents=True) + marker = workspace / "keep.txt" + marker.write_text("keep", encoding="utf-8") + runs_root = cleaner._workspace_root(workspace_base) + runs_root.parent.mkdir(parents=True) + try: + runs_root.symlink_to(outside, target_is_directory=True) + except OSError as exc: + pytest.skip(f"directory symlinks unavailable: {exc}") + + delete_result = _delete_workspace(workspace_base, run_id) + + assert delete_result.outcome is DeleteOutcome.REFUSED + assert marker.exists() + def test_refuses_detected_junction(self, workspace_base, monkeypatch): run_id = _run_id() path = _make_workspace(workspace_base, run_id) @@ -194,6 +213,23 @@ def test_refuses_detected_junction(self, workspace_base, monkeypatch): assert delete_result.outcome is DeleteOutcome.REFUSED assert path.exists() + def test_refuses_detected_workspace_root_junction( + self, workspace_base, monkeypatch + ): + run_id = _run_id() + path = _make_workspace(workspace_base, run_id) + runs_root = cleaner._workspace_root(workspace_base) + monkeypatch.setattr( + cleaner, + "_is_link_or_junction", + lambda candidate: candidate == runs_root, + ) + + delete_result = _delete_workspace(workspace_base, run_id) + + assert delete_result.outcome is DeleteOutcome.REFUSED + assert path.exists() + class TestPurgeTerminalWorkspaces: async def test_never_purges_active_runs(self, db_session_factory, workspace_base): From e8290756d12c8b783350760a47e8b048e32b7d1e Mon Sep 17 00:00:00 2001 From: Trung Minh Do Date: Tue, 18 Aug 2026 01:15:02 +0200 Subject: [PATCH 4/4] fix(automation): discover cleanup candidates on disk Filesystem discovery keeps orphaned workspaces subject to retention while the database only classifies discovered runs. Failed or refused deletions no longer consume the successful deletion budget, and the database session closes before deletion begins. --- openhands/automation/workspace_cleaner.py | 183 ++++++++++---- tests/test_workspace_cleaner.py | 292 ++++++++++++++++++++-- 2 files changed, 412 insertions(+), 63 deletions(-) diff --git a/openhands/automation/workspace_cleaner.py b/openhands/automation/workspace_cleaner.py index aec46b4..85397b0 100644 --- a/openhands/automation/workspace_cleaner.py +++ b/openhands/automation/workspace_cleaner.py @@ -8,8 +8,9 @@ import logging import os import shutil +import stat from dataclasses import dataclass -from datetime import timedelta +from datetime import UTC, datetime, timedelta from enum import Enum from pathlib import Path from uuid import UUID @@ -18,7 +19,7 @@ from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker from openhands.automation.models import AutomationRun, AutomationRunStatus -from openhands.automation.utils import utcnow +from openhands.automation.utils import ensure_utc, utcnow logger = logging.getLogger("automation.workspace_cleaner") @@ -75,6 +76,56 @@ def _workspace_path(workspace_base: str | Path, run_id: UUID) -> Path: return _workspace_root(workspace_base) / str(run_id) +def _scan_candidates(runs_root: Path) -> dict[UUID, float]: + """Find direct, canonical UUID workspace directories under ``runs_root``. + + The scan is intentionally filesystem-first: a workspace can outlive its + database row. Only direct child directories with canonical UUID names are + returned, and links/reparse points are rejected before their metadata is + read. Results are ordered oldest-first for deterministic cleanup. + """ + if _is_link_or_junction(runs_root): + logger.warning("Refusing linked workspace root during scan: %s", runs_root) + return {} + + try: + if not runs_root.is_dir(): + return {} + resolved_root = runs_root.resolve(strict=True) + candidates: list[tuple[UUID, float]] = [] + with os.scandir(runs_root) as entries: + for entry in entries: + candidate_path = Path(entry.path) + if entry.is_symlink() or _is_link_or_junction(candidate_path): + continue + try: + if not entry.is_dir(follow_symlinks=False): + continue + run_id = UUID(entry.name) + if str(run_id) != entry.name: + continue + resolved_path = candidate_path.resolve(strict=True) + if ( + resolved_path.parent != resolved_root + or not resolved_path.is_dir() + ): + continue + mtime = entry.stat(follow_symlinks=False).st_mtime + except (FileNotFoundError, OSError, ValueError): + # A candidate can disappear or become inaccessible while the + # directory is being scanned. It will be reconsidered later. + continue + candidates.append((run_id, mtime)) + except FileNotFoundError: + return {} + except OSError as exc: + logger.warning("Failed to scan workspace root %s: %s", runs_root, exc) + return {} + + candidates.sort(key=lambda item: (item[1], str(item[0]))) + return dict(candidates) + + def _dir_size(path: Path) -> int: total = 0 try: @@ -93,7 +144,32 @@ def _dir_size(path: Path) -> int: 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()) + if path.is_symlink() or (is_junction is not None and is_junction()): + return True + try: + attributes = getattr(path.stat(follow_symlinks=False), "st_file_attributes", 0) + except OSError: + return False + return bool(attributes & getattr(stat, "FILE_ATTRIBUTE_REPARSE_POINT", 0x400)) + + +def _is_expired( + completed_at: datetime | None, + mtime: float, + cutoff: datetime, +) -> bool: + """Return whether a terminal row or orphan is older than ``cutoff``. + + A terminal row without ``completed_at`` falls back to the workspace mtime; + this is also the age source for orphaned workspaces. Invalid filesystem + timestamps fail closed. + """ + try: + if completed_at is not None: + return ensure_utc(completed_at) < cutoff + return datetime.fromtimestamp(mtime, tz=UTC) < cutoff + except (OverflowError, OSError, ValueError): + return False def _delete_workspace( @@ -150,16 +226,18 @@ async def purge_terminal_workspaces( 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. + terminal state (COMPLETED, FAILED, CANCELLED, SKIPPED) older than the + retention cutoff are eligible. Terminal rows without ``completed_at`` use + workspace mtime as a safe fallback, as do orphaned workspaces. 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. + batch_size: Maximum number of successful workspace deletions per call. + Missing, refused, and failed deletion attempts do not consume the + limit, preventing one permanent failure from starving later + cleanup candidates. Returns: PurgeResult with counts of candidates, deletions, errors, and bytes freed. @@ -171,47 +249,60 @@ async def purge_terminal_workspaces( cutoff = utcnow() - timedelta(seconds=retention_seconds) result = PurgeResult() + candidates = await asyncio.to_thread( + _scan_candidates, _workspace_root(workspace_base) + ) + result.candidates_found = len(candidates) + + if candidates: + async with session_factory() as session: + stmt = select( + AutomationRun.id, + AutomationRun.status, + AutomationRun.completed_at, + ).where(AutomationRun.id.in_(list(candidates))) + rows = (await session.execute(stmt)).all() + by_id = { + run_id: (status, completed_at) for run_id, status, completed_at in rows + } + else: + by_id = {} + + for run_id, mtime in candidates.items(): + row = by_id.get(run_id) + if row is not None: + status, completed_at = row + if status not in TERMINAL_STATES: + continue + else: + completed_at = None - 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 not _is_expired(completed_at, mtime, cutoff): + continue - 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, - ) + delete_result = await asyncio.to_thread( + _delete_workspace, workspace_base, run_id + ) - attempted_existing = result.deleted + result.refused + result.errors - if attempted_existing >= batch_size: - break + if delete_result.outcome is DeleteOutcome.MISSING: + result.missing += 1 + continue + if delete_result.outcome is DeleteOutcome.REFUSED: + result.refused += 1 + continue + if delete_result.outcome is DeleteOutcome.ERROR: + result.errors += 1 + continue + + 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, + ) + if result.deleted >= batch_size: + break logger.info( "Purge complete: %d deleted, %d missing, %d refused, %d errors, " diff --git a/tests/test_workspace_cleaner.py b/tests/test_workspace_cleaner.py index 0f0b4d6..ed15146 100644 --- a/tests/test_workspace_cleaner.py +++ b/tests/test_workspace_cleaner.py @@ -1,5 +1,6 @@ """Tests for workspace purging of local-mode terminal runs.""" +import os import uuid from collections.abc import AsyncGenerator from datetime import datetime, timedelta @@ -26,6 +27,7 @@ PurgeResult, _delete_workspace, _dir_size, + _scan_candidates, _workspace_path, purge_terminal_workspaces, purger_loop, @@ -92,10 +94,18 @@ async def _create_run( await session.commit() -def _make_workspace(base: str, run_id: uuid.UUID, content: str = "data") -> Path: +def _make_workspace( + base: str, + run_id: uuid.UUID, + content: str = "data", + mtime: datetime | None = None, +) -> Path: path = _workspace_path(base, run_id) path.mkdir(parents=True, exist_ok=True) (path / "output.txt").write_text(content, encoding="utf-8") + if mtime is not None: + timestamp = mtime.timestamp() + os.utime(path, (timestamp, timestamp)) return path @@ -112,6 +122,60 @@ def test_native_and_forward_slash_bases_normalize_identically(self, tmp_path): assert native_path == forward_slash_path +class TestCandidateScan: + def test_skips_files_and_invalid_workspace_names(self, workspace_base): + runs_root = Path(workspace_base) / "automation-runs" + runs_root.mkdir(parents=True) + valid_id = _run_id() + (runs_root / str(valid_id)).mkdir() + (runs_root / str(_run_id())).write_text("not a directory", encoding="utf-8") + (runs_root / "not-a-uuid").mkdir() + uppercase_id = _run_id() + (runs_root / str(uppercase_id).upper()).mkdir() + + candidates = _scan_candidates(runs_root) + + assert list(candidates) == [valid_id] + + def test_skips_symlinked_workspace(self, workspace_base, tmp_path): + run_id = _run_id() + outside = tmp_path / "outside" + outside.mkdir() + marker = outside / "keep.txt" + marker.write_text("keep", encoding="utf-8") + runs_root = Path(workspace_base) / "automation-runs" + runs_root.mkdir(parents=True) + link = runs_root / str(run_id) + try: + link.symlink_to(outside, target_is_directory=True) + except OSError as exc: + pytest.skip(f"directory symlinks unavailable: {exc}") + + candidates = _scan_candidates(runs_root) + + assert candidates == {} + assert marker.exists() + + def test_skips_detected_reparse_point(self, workspace_base, monkeypatch): + run_id = _run_id() + runs_root = Path(workspace_base) / "automation-runs" + path = runs_root / str(run_id) + path.mkdir(parents=True) + monkeypatch.setattr( + cleaner, + "_is_link_or_junction", + lambda candidate: candidate == path, + ) + + assert _scan_candidates(runs_root) == {} + + def test_empty_runs_root_is_safe(self, workspace_base): + runs_root = Path(workspace_base) / "automation-runs" + runs_root.mkdir(parents=True) + + assert _scan_candidates(runs_root) == {} + + class TestDirSize: def test_empty_dir(self, workspace_base): path = _workspace_path(workspace_base, _run_id()) @@ -261,7 +325,7 @@ async def test_never_purges_active_runs(self, db_session_factory, workspace_base db_session_factory, workspace_base, retention_seconds=0, batch_size=10 ) - assert result.candidates_found == 0 + assert result.candidates_found == 2 assert result.deleted == 0 assert pending_path.exists() assert running_path.exists() @@ -285,7 +349,7 @@ async def test_respects_retention_cutoff(self, db_session_factory, workspace_bas db_session_factory, workspace_base, retention_seconds=3600, batch_size=10 ) - assert result.candidates_found == 0 + assert result.candidates_found == 1 assert result.deleted == 0 assert path.exists() @@ -343,10 +407,10 @@ async def test_all_terminal_states(self, db_session_factory, workspace_base): assert result.deleted == 4 assert result.errors == 0 - async def test_tolerates_missing_workspace( + async def test_missing_database_workspace_is_not_a_filesystem_candidate( self, db_session_factory, workspace_base ): - """Runs without workspace directories must not cause errors.""" + """Rows without directories are invisible to filesystem-first discovery.""" auto_id = uuid.uuid4() await _create_automation(db_session_factory, auto_id) @@ -363,9 +427,9 @@ async def test_tolerates_missing_workspace( db_session_factory, workspace_base, retention_seconds=3600, batch_size=10 ) - assert result.candidates_found == 1 + assert result.candidates_found == 0 assert result.deleted == 0 - assert result.missing == 1 + assert result.missing == 0 assert result.errors == 0 async def test_respects_batch_size(self, db_session_factory, workspace_base): @@ -389,7 +453,7 @@ async def test_respects_batch_size(self, db_session_factory, workspace_base): db_session_factory, workspace_base, retention_seconds=3600, batch_size=3 ) - assert result.candidates_found == 3 + assert result.candidates_found == 5 assert result.deleted == 3 remaining = list((Path(workspace_base) / "automation-runs").iterdir()) assert len(remaining) == 2 @@ -404,30 +468,224 @@ async def test_empty_database(self, db_session_factory, workspace_base): assert result.deleted == 0 assert result.errors == 0 - async def test_skips_runs_without_completed_at( + async def test_purges_old_orphan_workspace( self, db_session_factory, workspace_base ): - """Terminal runs without completed_at are skipped.""" + """An old valid workspace without a database row is an orphan.""" + run_id = _run_id() + path = _make_workspace( + workspace_base, + run_id, + mtime=utcnow() - timedelta(days=30), + ) + + result = await purge_terminal_workspaces( + db_session_factory, workspace_base, retention_seconds=3600, batch_size=10 + ) + + assert result.candidates_found == 1 + assert result.deleted == 1 + assert not path.exists() + + async def test_keeps_young_orphan_workspace( + self, db_session_factory, workspace_base + ): + """A young orphan remains until its filesystem retention expires.""" + run_id = _run_id() + path = _make_workspace(workspace_base, run_id) + + result = await purge_terminal_workspaces( + db_session_factory, workspace_base, retention_seconds=3600, batch_size=10 + ) + + assert result.candidates_found == 1 + assert result.deleted == 0 + assert path.exists() + + async def test_classifies_orphan_and_database_rows_together( + self, db_session_factory, workspace_base + ): + """Filesystem candidates are classified together by one DB snapshot.""" auto_id = uuid.uuid4() await _create_automation(db_session_factory, auto_id) + old_time = utcnow() - timedelta(days=30) + + orphan_id = _run_id() + terminal_id = _run_id() + pending_id = _run_id() + await _create_run( + db_session_factory, + terminal_id, + auto_id, + AutomationRunStatus.COMPLETED, + completed_at=old_time, + ) + await _create_run( + db_session_factory, + pending_id, + auto_id, + AutomationRunStatus.PENDING, + completed_at=old_time, + ) + orphan_path = _make_workspace(workspace_base, orphan_id, mtime=old_time) + terminal_path = _make_workspace(workspace_base, terminal_id) + pending_path = _make_workspace(workspace_base, pending_id) + result = await purge_terminal_workspaces( + db_session_factory, workspace_base, retention_seconds=3600, batch_size=10 + ) + + assert result.candidates_found == 3 + assert result.deleted == 2 + assert not orphan_path.exists() + assert not terminal_path.exists() + assert pending_path.exists() + + async def test_uses_one_batched_database_lookup( + self, db_session_factory, workspace_base, monkeypatch + ): + auto_id = uuid.uuid4() + await _create_automation(db_session_factory, auto_id) + old_time = utcnow() - timedelta(days=30) + for _ in range(3): + run_id = _run_id() + await _create_run( + db_session_factory, + run_id, + auto_id, + AutomationRunStatus.COMPLETED, + completed_at=old_time, + ) + _make_workspace(workspace_base, run_id) + + execute = AsyncSession.execute + execute_calls = 0 + + async def spy_execute(session, statement, *args, **kwargs): + nonlocal execute_calls + execute_calls += 1 + return await execute(session, statement, *args, **kwargs) + + monkeypatch.setattr(AsyncSession, "execute", spy_execute) + await purge_terminal_workspaces( + db_session_factory, workspace_base, retention_seconds=3600, batch_size=10 + ) + + assert execute_calls == 1 + + async def test_permanent_delete_error_does_not_starve_newer_workspace( + self, db_session_factory, workspace_base, monkeypatch + ): + """A failed old candidate cannot consume the whole batch budget.""" + auto_id = uuid.uuid4() + await _create_automation(db_session_factory, auto_id) + old_time = utcnow() - timedelta(days=30) + older_id = _run_id() + newer_id = _run_id() + await _create_run( + db_session_factory, + older_id, + auto_id, + AutomationRunStatus.COMPLETED, + completed_at=old_time - timedelta(seconds=1), + ) + await _create_run( + db_session_factory, + newer_id, + auto_id, + AutomationRunStatus.COMPLETED, + completed_at=old_time, + ) + older_path = _make_workspace( + workspace_base, + older_id, + mtime=old_time - timedelta(seconds=1), + ) + newer_path = _make_workspace(workspace_base, newer_id, mtime=old_time) + real_rmtree = cleaner.shutil.rmtree + + def fail_old(path, *args, **kwargs): + if Path(path) == older_path: + raise PermissionError("locked") + return real_rmtree(path, *args, **kwargs) + + monkeypatch.setattr(cleaner.shutil, "rmtree", fail_old) + + first_result = await purge_terminal_workspaces( + db_session_factory, workspace_base, retention_seconds=3600, batch_size=1 + ) + second_result = await purge_terminal_workspaces( + db_session_factory, workspace_base, retention_seconds=3600, batch_size=1 + ) + + assert first_result.errors == 1 + assert first_result.deleted == 1 + assert second_result.errors == 1 + assert second_result.deleted == 0 + assert older_path.exists() + assert not newer_path.exists() + + async def test_vanished_candidate_is_treated_as_missing( + self, db_session_factory, workspace_base, monkeypatch + ): + auto_id = uuid.uuid4() + await _create_automation(db_session_factory, auto_id) run_id = _run_id() await _create_run( db_session_factory, run_id, auto_id, - AutomationRunStatus.CANCELLED, - completed_at=None, + AutomationRunStatus.COMPLETED, + completed_at=utcnow() - timedelta(days=30), ) path = _make_workspace(workspace_base, run_id) + real_delete = cleaner._delete_workspace + + def disappear_before_delete(base, candidate_id): + if path.exists(): + real_rmtree = cleaner.shutil.rmtree + real_rmtree(path) + return real_delete(base, candidate_id) + + monkeypatch.setattr(cleaner, "_delete_workspace", disappear_before_delete) result = await purge_terminal_workspaces( - db_session_factory, workspace_base, retention_seconds=0, batch_size=10 + db_session_factory, workspace_base, retention_seconds=3600, batch_size=1 ) - assert result.candidates_found == 0 + assert result.candidates_found == 1 + assert result.missing == 1 assert result.deleted == 0 - assert path.exists() + assert result.errors == 0 + + async def test_terminal_run_without_completed_at_uses_workspace_mtime( + self, db_session_factory, workspace_base + ): + """Missing completed_at falls back to the workspace mtime.""" + auto_id = uuid.uuid4() + await _create_automation(db_session_factory, auto_id) + + run_id = _run_id() + await _create_run( + db_session_factory, + run_id, + auto_id, + AutomationRunStatus.CANCELLED, + completed_at=None, + ) + path = _make_workspace( + workspace_base, + run_id, + mtime=utcnow() - timedelta(days=30), + ) + + result = await purge_terminal_workspaces( + db_session_factory, workspace_base, retention_seconds=3600, batch_size=10 + ) + + assert result.candidates_found == 1 + assert result.deleted == 1 + assert not path.exists() async def test_workspace_base_expansion( self, db_session_factory, tmp_path, monkeypatch @@ -490,8 +748,8 @@ async def test_missing_old_row_does_not_starve_existing_workspace( db_session_factory, workspace_base, retention_seconds=3600, batch_size=1 ) - assert result.candidates_found == 2 - assert result.missing == 1 + assert result.candidates_found == 1 + assert result.missing == 0 assert result.deleted == 1 assert not existing_path.exists()