refactor(ops): address re-review findings for cleanup commands

- Use model_copy(update=...) in test helpers so Pydantic ge=
  validators are enforced during tests, not just env vars (NEW-1)
- Promote _extract_plan_id_from_sandbox to public staticmethod
  and reuse in CLI active-plan detection to eliminate duplicate
  plan-ID parsing logic (NEW-2)
- Cache _get_sandbox_dirs result for service instance lifetime
  and have CLI active-plan detection use the cached listing so
  /tmp is iterated only once per invocation (NEW-3)
This commit is contained in:
khyari hamza
2026-02-19 14:20:16 +00:00
parent f4c088bae8
commit cea3bad758
4 changed files with 59 additions and 46 deletions
@@ -24,12 +24,16 @@ from cleveragents.config.settings import Settings
def _make_settings(**overrides: object) -> Settings:
"""Create a Settings instance with optional field overrides."""
"""Create a Settings instance with optional field overrides.
Uses ``model_copy(update=...)`` so that Pydantic validators
(including ``ge=`` bounds on retention fields) are enforced.
"""
Settings._instance = None
settings = Settings()
for key, value in overrides.items():
setattr(settings, key, value)
return settings
base = Settings()
if overrides:
return base.model_copy(update=overrides)
return base
def _create_sandbox_dir(
+9 -6
View File
@@ -26,13 +26,16 @@ from cleveragents.config.settings import Settings
def _make_settings(**overrides: object) -> Settings:
"""Create a Settings instance with optional field overrides."""
# Reset the singleton to get fresh defaults
"""Create a Settings instance with optional field overrides.
Uses ``model_copy(update=...)`` so that Pydantic validators
(including ``ge=`` bounds on retention fields) are enforced.
"""
Settings._instance = None
settings = Settings()
for key, value in overrides.items():
setattr(settings, key, value)
return settings
base = Settings()
if overrides:
return base.model_copy(update=overrides)
return base
def _create_sandbox_dir(
@@ -123,6 +123,7 @@ class CleanupService:
raise TypeError("settings must be a Settings instance")
self._settings = settings
self._active_plan_ids: frozenset[str] = active_plan_ids or frozenset()
self._sandbox_dirs_cache: list[Path] | None = None
# ── Public API ────────────────────────────────────────────────
@@ -163,15 +164,24 @@ class CleanupService:
# ── Sandbox cleanup ───────────────────────────────────────────
def _get_sandbox_dirs(self) -> list[Path]:
"""Find sandbox directories in the system temp directory."""
"""Find sandbox directories in the system temp directory.
Results are cached for the lifetime of the service instance so
that ``scan()`` and ``purge()`` (and the CLI active-plan check)
do not redundantly iterate ``/tmp``.
"""
if self._sandbox_dirs_cache is not None:
return self._sandbox_dirs_cache
tmp = Path(tempfile.gettempdir())
if not tmp.exists():
return []
self._sandbox_dirs_cache = []
return self._sandbox_dirs_cache
dirs: list[Path] = []
try:
entries = list(tmp.iterdir())
except OSError:
return []
self._sandbox_dirs_cache = []
return self._sandbox_dirs_cache
for p in entries:
try:
is_dir = p.is_dir()
@@ -181,13 +191,19 @@ class CleanupService:
p.name.startswith(pfx) for pfx in ("ca-sandbox-", "ca-cow-sandbox-")
):
dirs.append(p)
self._sandbox_dirs_cache = dirs
return dirs
def _extract_plan_id_from_sandbox(self, path: Path) -> str | None:
@staticmethod
def extract_plan_id_from_sandbox(path: Path) -> str | None:
"""Extract plan_id from a sandbox directory name.
Handles both standard (``ca-sandbox-<plan_id>-<random>``) and
copy-on-write (``ca-cow-sandbox-<plan_id>-<random>``) formats.
This is a public static method so that callers (e.g. the CLI
active-plan detection) can reuse the parsing logic without
constructing a full :class:`CleanupService`.
"""
name = path.name
for prefix in ("ca-cow-sandbox-", "ca-sandbox-"):
@@ -213,7 +229,7 @@ class CleanupService:
dirs = self._get_sandbox_dirs()
for d in dirs:
report.sandboxes.scanned += 1
plan_id = self._extract_plan_id_from_sandbox(d)
plan_id = self.extract_plan_id_from_sandbox(d)
if plan_id and plan_id in self._active_plan_ids:
report.sandboxes.skipped += 1
report.sandboxes.skipped_reasons.append(
@@ -235,7 +251,7 @@ class CleanupService:
dirs = self._get_sandbox_dirs()
for d in dirs:
report.sandboxes.scanned += 1
plan_id = self._extract_plan_id_from_sandbox(d)
plan_id = self.extract_plan_id_from_sandbox(d)
if plan_id and plan_id in self._active_plan_ids:
report.sandboxes.skipped += 1
report.sandboxes.skipped_reasons.append(
+19 -29
View File
@@ -6,7 +6,6 @@ stale sandboxes, checkpoints, sessions, logs, and backups.
from __future__ import annotations
from pathlib import Path
from typing import TYPE_CHECKING, Annotated
import typer
@@ -24,39 +23,24 @@ app = typer.Typer(
)
def _get_active_plan_ids() -> frozenset[str]:
"""Best-effort query for currently active plan IDs.
def _detect_active_plan_ids(service: CleanupService) -> frozenset[str]:
"""Best-effort detection of currently active plan IDs.
Scans sandbox directory names for embedded plan IDs that have
recent modification times (< 1 hour), treating them as likely
active. Falls back to an empty set with a warning when the
temp directory cannot be read.
Reuses the service's cached sandbox directory listing and its
:meth:`~CleanupService.extract_plan_id_from_sandbox` parser so
that ``/tmp`` is only iterated once per CLI invocation.
A future version can query the plan lifecycle service directly
once the Container is available at CLI startup.
Sandboxes modified within the last hour are assumed to belong
to running plans. Falls back to an empty set with a warning
when the temp directory cannot be read.
"""
import tempfile
import time
try:
tmp = Path(tempfile.gettempdir())
if not tmp.exists():
return frozenset()
active: set[str] = set()
one_hour_ago = time.time() - 3600
for p in tmp.iterdir():
if not p.is_dir():
continue
name = p.name
plan_id: str | None = None
if name.startswith("ca-sandbox-"):
parts = name[len("ca-sandbox-") :].rsplit("-", 1)
if parts:
plan_id = parts[0]
elif name.startswith("ca-cow-sandbox-"):
parts = name[len("ca-cow-sandbox-") :].rsplit("-", 1)
if parts:
plan_id = parts[0]
for p in service._get_sandbox_dirs():
plan_id = service.extract_plan_id_from_sandbox(p)
if plan_id:
try:
mtime = p.stat().st_mtime
@@ -75,13 +59,19 @@ def _get_active_plan_ids() -> frozenset[str]:
def _get_cleanup_service() -> CleanupService:
"""Build a :class:`CleanupService` from current settings."""
"""Build a :class:`CleanupService` from current settings.
Constructs the service, detects active plans from the cached
sandbox listing, then injects the active set before returning.
This ensures ``/tmp`` is iterated only once.
"""
from cleveragents.application.services.cleanup_service import CleanupService
from cleveragents.config.settings import get_settings
settings = get_settings()
active_plan_ids = _get_active_plan_ids()
return CleanupService(settings=settings, active_plan_ids=active_plan_ids)
service = CleanupService(settings=settings)
service._active_plan_ids = _detect_active_plan_ids(service)
return service
@app.command(name="scan")