diff --git a/README.md b/README.md index 6d0fb52..502460e 100644 --- a/README.md +++ b/README.md @@ -747,15 +747,34 @@ history writer to persist redacted command metadata, timing, exit status, project context, and a pointer to the raw log file. History writes should be best-effort and should not fail the user command when an index cannot be updated. -High-frequency tools can set `base_cli.App(max_log_files=)` to keep at -most that many default persistent log files across the owner's run bundles. +Durable applications use a bounded complete-bundle policy by default, so +retention never splits a run's metadata, log, and temporary diagnostics. +High-frequency tools can tune it explicitly: + +```python +app = base_cli.App( + name="workspace-tools", + retention=base_cli.RetentionPolicy( + max_bundles=20, + max_age_seconds=30 * 24 * 60 * 60, + max_total_bytes=512 * 1024 * 1024, + ), +) +``` + Retention runs during startup after the current run's default log file is -resolved, and the current run's log file is never pruned. The policy is skipped -for `ctx.dry_run`, -`log_to_file=False`, and explicit `--log-file` paths so no-durable-write modes -and caller-selected log locations stay under caller control. Use this as a -small guardrail for busy local tools; an application can provide broader -maintenance commands for caches, logs, and retained temp files. +resolved. The active invocation, inherited parent bundle, and bundles marked +`preserve` (including `--keep-temp`) are never removed. A stale `running` +bundle is eligible for crash recovery only when an age bound is configured; +without one it is retained for diagnosis. The metadata and retention index are +written with same-filesystem temporary files, flush/sync, and atomic +replacement, and concurrent pruners serialize through a sidecar lock. + +The policy is skipped for `ctx.dry_run`, `log_to_file=False`, and explicit +`--log-file` paths so no-durable-write modes and caller-selected log locations +stay under caller control. The original `max_log_files=` option remains +available as a compatibility per-file policy; new applications should use +`RetentionPolicy`. Logs use a stable, human-readable shape: diff --git a/lib/python/base_cli/__init__.py b/lib/python/base_cli/__init__.py index d380320..ff054c6 100644 --- a/lib/python/base_cli/__init__.py +++ b/lib/python/base_cli/__init__.py @@ -117,7 +117,7 @@ def _resolve_version() -> str: UserConfigLoader, WorkspaceRootResolver, ) -from .runtime import RuntimeLayout +from .runtime import RetentionPolicy, RuntimeLayout from .typer import TyperAdapter, attach_typer, get_typer_command __all__ = [ @@ -195,6 +195,7 @@ def _resolve_version() -> str: "ProjectDiscovery", "RECORD_SCHEMAS", "RuntimeLayout", + "RetentionPolicy", "RuntimeResolver", "is_terminal", "output_format_choices", diff --git a/lib/python/base_cli/_lifecycle.py b/lib/python/base_cli/_lifecycle.py index 2abbcbc..a886b25 100644 --- a/lib/python/base_cli/_lifecycle.py +++ b/lib/python/base_cli/_lifecycle.py @@ -9,6 +9,7 @@ from .context import Context from .exit_codes import ExitCode from .history import format_timestamp +from ._runtime import refresh_run_bundle_index @dataclass(frozen=True) @@ -57,6 +58,9 @@ def finish( } ) write_private_json(self.context._run_metadata_path, metadata) + owner_root = self.context.owner_root + if owner_root is not None: + refresh_run_bundle_index(owner_root / "runs", logger=self.context.log) def _existing_metadata(self) -> dict[str, Any]: path = self.context._run_metadata_path @@ -103,6 +107,10 @@ def _metadata(self, *, status: str) -> dict[str, Any]: "project_root": str(context.project_root) if context.project_root else None, "manifest": str(context.manifest_path) if context.manifest_path else None, "workspace_root": str(context.workspace_root) if context.workspace_root else None, + # ``--keep-temp`` is an explicit request to retain diagnostics; + # retention therefore protects the complete invocation bundle, + # not only its temporary directory. + "preserve": context.keep_temp, } diff --git a/lib/python/base_cli/_private_files.py b/lib/python/base_cli/_private_files.py index 44133d5..605b946 100644 --- a/lib/python/base_cli/_private_files.py +++ b/lib/python/base_cli/_private_files.py @@ -4,6 +4,7 @@ import json import os +import secrets from collections.abc import Mapping from pathlib import Path from typing import Any @@ -32,20 +33,148 @@ def restrict_directory(path: Path) -> None: def write_private_json(path: Path, value: Mapping[str, Any]) -> None: - """Write a JSON mapping with owner-only permissions from the moment it is created.""" + """Atomically write a private JSON mapping. + The temporary file is created beside the destination, written and synced + before it is replaced. A failed serialization, write, or replacement + therefore leaves the previous snapshot intact instead of exposing a + truncated metadata/index file. On POSIX, directory-relative operations + also prevent a swapped ancestor from redirecting the replacement through + a symlink. + """ + + path = Path(path) path.parent.mkdir(parents=True, exist_ok=True) - fd = os.open(path, os.O_WRONLY | os.O_CREAT | os.O_TRUNC, PRIVATE_FILE_MODE) + parent_fd: int | None = None + temporary_name: str | None = None + fd: int | None = None try: + parent_fd = _open_parent_directory(path.parent) + for _ in range(32): + candidate = f".{path.name}.{secrets.token_hex(12)}.tmp" + try: + if parent_fd is not None and _supports_dir_fd_open(): + fd = os.open( + candidate, + os.O_WRONLY | os.O_CREAT | os.O_EXCL, + PRIVATE_FILE_MODE, + dir_fd=parent_fd, + ) + else: + fd = os.open( + path.parent / candidate, + os.O_WRONLY | os.O_CREAT | os.O_EXCL, + PRIVATE_FILE_MODE, + ) + temporary_name = candidate + break + except FileExistsError: + continue + if fd is None or temporary_name is None: + raise OSError("unable to allocate a private JSON temporary file") + fchmod = getattr(os, "fchmod", None) if os.name != "nt" and fchmod is not None: fchmod(fd, PRIVATE_FILE_MODE) - stream = os.fdopen(fd, "w", encoding="utf-8") - fd = -1 - with stream: + with os.fdopen(fd, "w", encoding="utf-8") as stream: + fd = None json.dump(value, stream, sort_keys=True) stream.write("\n") + stream.flush() + os.fsync(stream.fileno()) + + # Refuse an existing symlink as the destination. Replacing a regular + # file is safe and atomic; replacing a symlink entry would be safe from + # following it, but rejecting it makes an unexpected redirection + # explicit to callers. + if parent_fd is not None and _supports_dir_fd_stat(): + try: + existing = os.stat(path.name, dir_fd=parent_fd, follow_symlinks=False) + except FileNotFoundError: + existing = None + if existing is not None and _is_symlink_mode(existing.st_mode): + raise OSError(f"refusing to replace symlink '{path}'") + _atomic_replace_relative(parent_fd, temporary_name, path.name) + else: + if path.is_symlink(): + raise OSError(f"refusing to replace symlink '{path}'") + os.replace(path.parent / temporary_name, path) + temporary_name = None + if parent_fd is not None: + _sync_directory(parent_fd) + restrict_file(path) finally: - if fd != -1: + if fd is not None: os.close(fd) - restrict_file(path) + if temporary_name is not None: + try: + if parent_fd is not None and _supports_dir_fd_unlink(): + os.unlink(temporary_name, dir_fd=parent_fd) + else: + (path.parent / temporary_name).unlink() + except OSError: + pass + if parent_fd is not None: + os.close(parent_fd) + + +def _supports_dir_fd_open() -> bool: + return os.open in os.supports_dir_fd + + +def _supports_dir_fd_stat() -> bool: + return os.stat in os.supports_dir_fd and os.stat in os.supports_follow_symlinks + + +def _supports_dir_fd_unlink() -> bool: + return os.unlink in os.supports_dir_fd + + +def _is_symlink_mode(mode: int) -> bool: + # Avoid importing stat in the hot path and keep this helper usable on + # platforms where the POSIX mode constants are still available. + return (mode & 0o170000) == 0o120000 + + +def _open_parent_directory(path: Path) -> int | None: + """Open a parent directory without following path components on POSIX.""" + + if os.name == "nt" or not hasattr(os, "O_NOFOLLOW") or not hasattr(os, "O_DIRECTORY"): + return None + # Resolve the platform's standard compatibility links (for example + # macOS's /var -> /private/var), then bind each real component with + # O_NOFOLLOW so a later ancestor swap cannot redirect the write. + absolute = path.resolve(strict=True) + descriptor = os.open( + absolute.anchor, + os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW | getattr(os, "O_CLOEXEC", 0), + ) + try: + for component in absolute.relative_to(Path(absolute.anchor)).parts: + next_descriptor = os.open( + component, + os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW | getattr(os, "O_CLOEXEC", 0), + dir_fd=descriptor, + ) + os.close(descriptor) + descriptor = next_descriptor + return descriptor + except BaseException: + os.close(descriptor) + raise + + +def _atomic_replace_relative(parent_fd: int, source: str, destination: str) -> None: + if os.rename in os.supports_dir_fd: + os.rename(source, destination, src_dir_fd=parent_fd, dst_dir_fd=parent_fd) + return + os.replace(source, destination) + + +def _sync_directory(parent_fd: int) -> None: + try: + os.fsync(parent_fd) + except OSError: + # Directory fsync is not available on all supported filesystems and + # platforms. The file itself was still flushed before replacement. + pass diff --git a/lib/python/base_cli/_runtime.py b/lib/python/base_cli/_runtime.py index d117543..c373b6e 100644 --- a/lib/python/base_cli/_runtime.py +++ b/lib/python/base_cli/_runtime.py @@ -4,14 +4,35 @@ import logging import os import stat +import time +from contextlib import contextmanager +from datetime import datetime, timezone +from typing import Any, Iterable, Iterator from pathlib import Path -from ._private_files import PRIVATE_DIRECTORY_MODE, restrict_directory, write_private_json +try: # pragma: no cover - platform branch + import fcntl as _fcntl +except ImportError: # pragma: no cover - Windows + _fcntl = None # type: ignore[assignment] + +try: # pragma: no cover - platform branch + import msvcrt as _msvcrt +except ImportError: # pragma: no cover - POSIX + _msvcrt = None # type: ignore[assignment] + +from ._private_files import ( + PRIVATE_DIRECTORY_MODE, + restrict_directory, + restrict_file, + write_private_json, +) from .paths import runtime_run_directory_name, runtime_slug -from .runtime import RuntimeLayout +from .runtime import RetentionPolicy, RuntimeLayout _LOG_INDEX_NAME = ".base-cli-log-index.json" +_RUN_INDEX_NAME = ".base-cli-run-index.json" +_RUN_LOCK_NAME = ".base-cli-run-index.lock" class RuntimeDirectoryError(RuntimeError): @@ -198,6 +219,18 @@ def prune_log_files( max_log_files: int, logger: logging.Logger, ) -> None: + # Modern runtimes have one ``run.json`` per direct child and must be + # pruned as complete bundles. Keep this legacy helper available for old + # flat log directories, but never let it split a metadata-backed bundle. + try: + if any( + (child / "run.json").is_file() + for child in log_dir.iterdir() + if child.is_dir() and not child.is_symlink() + ): + return + except OSError: + return index_path = log_dir / _LOG_INDEX_NAME tracked = _read_log_index(index_path) if tracked is None: @@ -224,6 +257,396 @@ def prune_log_files( logger.debug("Could not update log retention index '%s': %s", index_path, exc) +def prune_run_bundles( + runs_root: Path, + current_run_root: Path | None = None, + *, + policy: RetentionPolicy | None = None, + max_bundles: int | None = None, + max_age_seconds: float | None = None, + max_total_bytes: int | None = None, + protected_run_roots: Iterable[Path] = (), + logger: logging.Logger | None = None, + now: float | None = None, +) -> None: + """Prune complete invocation bundles under ``runs_root``. + + A bundle is a direct child directory containing valid ``run.json`` + metadata in a terminal state. The metadata file and the retention index + are both updated atomically. We intentionally scan the filesystem while + holding a sidecar lock rather than trusting the index for deletion: a + damaged or stale index can never cause an unrelated path to be removed. + + Running bundles are protected while active. A running bundle is eligible + for crash recovery only when an age bound is configured and it is older + than that bound; the current and explicitly protected roots remain safe. + """ + + log = logger or logging.getLogger(__name__) + if policy is not None: + if any(value is not None for value in (max_bundles, max_age_seconds, max_total_bytes)): + raise ValueError("pass either policy or individual retention bounds, not both") + max_bundles = policy.max_bundles + max_age_seconds = policy.max_age_seconds + max_total_bytes = policy.max_total_bytes + effective = RetentionPolicy(max_bundles, max_age_seconds, max_total_bytes) + if effective.max_bundles is None and effective.max_age_seconds is None and effective.max_total_bytes is None: + return + + runs_root = Path(runs_root) + if not runs_root.exists() or runs_root.is_symlink(): + return + protected = {_safe_resolved_path(path) for path in protected_run_roots} + if current_run_root is not None: + protected.add(_safe_resolved_path(current_run_root)) + clock = time.time() if now is None else now + + try: + with _retention_lock(runs_root): + bundles = _discover_run_bundles( + runs_root, + protected=protected, + max_age_seconds=effective.max_age_seconds, + now=clock, + ) + _apply_bundle_retention( + runs_root, + bundles, + policy=effective, + protected=protected, + logger=log, + now=clock, + reserved_active_bundles=1 if current_run_root is not None else 0, + ) + _write_run_index( + runs_root, + bundles, + log, + current_run_root=current_run_root, + now=clock, + ) + except (OSError, RuntimeError) as exc: + # Retention is maintenance. An unavailable lock or a transient + # filesystem failure must not turn an otherwise valid invocation into + # a command failure, and the old bundle remains untouched. + log.warning("Could not update run bundle retention under '%s': %s", runs_root, exc) + + +def refresh_run_bundle_index( + runs_root: Path, + *, + logger: logging.Logger | None = None, +) -> None: + """Refresh the diagnostic bundle index after a run becomes terminal.""" + + log = logger or logging.getLogger(__name__) + runs_root = Path(runs_root) + if not runs_root.exists() or runs_root.is_symlink(): + return + try: + with _retention_lock(runs_root): + bundles = _discover_run_bundles( + runs_root, + protected=set(), + max_age_seconds=None, + now=time.time(), + ) + _write_run_index(runs_root, bundles, log) + except (OSError, RuntimeError) as exc: + log.debug("Could not refresh run bundle index under '%s': %s", runs_root, exc) + + +def _discover_run_bundles( + runs_root: Path, + *, + protected: set[Path], + max_age_seconds: float | None, + now: float, +) -> list[dict[str, Any]]: + bundles: list[dict[str, Any]] = [] + try: + children = sorted(runs_root.iterdir(), key=lambda path: path.name) + except OSError: + return bundles + for child in children: + if child.name.startswith(".") or child.is_symlink() or not child.is_dir(): + continue + metadata = _read_bundle_metadata(child) + if metadata is None: + # Partial startup directories are deliberately not considered + # complete. They can be diagnosed or cleaned by a consumer's + # explicit maintenance command without retention guessing. + continue + status = str(metadata.get("status", "")) + started_at = _timestamp_to_epoch(metadata.get("started_at")) + if started_at is None: + try: + started_at = child.stat().st_mtime + except OSError: + continue + age = max(0.0, now - started_at) + running = status == "running" + stale_running = running and max_age_seconds is not None and age >= max_age_seconds + if running and not stale_running: + continue + if status not in {"running", "ok", "error"}: + continue + resolved = _safe_resolved_path(child) + try: + size = _bundle_size(child) + except OSError: + continue + retention_metadata = metadata.get("retention") + preserve = bool(metadata.get("preserve")) or ( + isinstance(retention_metadata, dict) and retention_metadata.get("preserve") is True + ) + bundles.append( + { + "path": child, + "resolved": resolved, + "run_id": metadata.get("run_id"), + "status": status, + "started_at": started_at, + "age": age, + "size": size, + "preserve": preserve, + "protected": resolved in protected, + } + ) + bundles.sort(key=lambda bundle: (float(bundle["started_at"]), str(bundle["path"]))) + return bundles + + +def _apply_bundle_retention( + runs_root: Path, + bundles: list[dict[str, Any]], + *, + policy: RetentionPolicy, + protected: set[Path], + logger: logging.Logger, + now: float, + reserved_active_bundles: int, +) -> None: + del now # retained for a stable extension point in policy implementations + removable = [ + bundle + for bundle in bundles + if not bool(bundle["protected"]) + and not bool(bundle["preserve"]) + and _safe_resolved_path(bundle["path"]) not in protected + ] + removable.sort(key=lambda bundle: (float(bundle["started_at"]), str(bundle["path"]))) + + def remove(bundle: dict[str, Any]) -> bool: + path = Path(bundle["path"]) + try: + _remove_run_bundle(runs_root, path) + except OSError as exc: + logger.warning("Could not prune run bundle '%s': %s", path, exc) + removable.remove(bundle) + return False + bundles.remove(bundle) + removable.remove(bundle) + return True + + if policy.max_age_seconds is not None: + for bundle in list(removable): + if float(bundle["age"]) >= policy.max_age_seconds: + remove(bundle) + + if policy.max_bundles is not None: + while len(bundles) + reserved_active_bundles > policy.max_bundles and removable: + remove(removable[0]) + + if policy.max_total_bytes is not None: + total = sum(int(bundle["size"]) for bundle in bundles) + while total > policy.max_total_bytes and removable: + candidate = removable[0] + if remove(candidate): + total -= int(candidate["size"]) + + +def _write_run_index( + runs_root: Path, + bundles: list[dict[str, Any]], + logger: logging.Logger, + *, + current_run_root: Path | None = None, + now: float | None = None, +) -> None: + indexed = list(bundles) + if current_run_root is not None and current_run_root.exists(): + current_resolved = _safe_resolved_path(current_run_root) + if not any(_safe_resolved_path(Path(bundle["path"])) == current_resolved for bundle in indexed): + indexed.append( + { + "path": current_run_root, + "run_id": current_run_root.name, + "status": "running", + "started_at": time.time() if now is None else now, + "size": _bundle_size(current_run_root), + "preserve": False, + } + ) + index_path = runs_root / _RUN_INDEX_NAME + payload = { + "version": 1, + "bundles": [ + { + "path": str(bundle["path"]), + "run_id": bundle["run_id"], + "status": bundle["status"], + "started_at": bundle["started_at"], + "size": bundle["size"], + "preserve": bool(bundle["preserve"]), + } + for bundle in indexed + ], + } + try: + write_private_json(index_path, payload) + except (OSError, TypeError, ValueError) as exc: + logger.debug("Could not update run bundle retention index '%s': %s", index_path, exc) + + +def _read_bundle_metadata(path: Path) -> dict[str, object] | None: + metadata_path = path / "run.json" + try: + metadata_stat = metadata_path.stat() + if not stat.S_ISREG(metadata_stat.st_mode) or metadata_path.is_symlink(): + return None + payload = json.loads(metadata_path.read_text(encoding="utf-8")) + except (OSError, UnicodeDecodeError, json.JSONDecodeError): + return None + return payload if isinstance(payload, dict) else None + + +def _timestamp_to_epoch(value: object) -> float | None: + if not isinstance(value, str): + return None + try: + parsed = datetime.fromisoformat(value.replace("Z", "+00:00")) + except ValueError: + return None + if parsed.tzinfo is None: + parsed = parsed.replace(tzinfo=timezone.utc) + return parsed.timestamp() + + +def _bundle_size(path: Path) -> int: + total = 0 + for root, directories, files in os.walk(path, topdown=True, followlinks=False): + for name in [*directories, *files]: + child = Path(root) / name + try: + total += child.lstat().st_size + except OSError: + continue + try: + total += path.lstat().st_size + except OSError: + pass + return total + + +def _remove_run_bundle(runs_root: Path, path: Path) -> None: + """Remove one direct child using descriptor-relative, no-follow operations.""" + + root_fd = _open_directory_nofollow(runs_root) + try: + candidate = Path(path).name + if Path(path).parent.resolve() != runs_root.resolve(): + raise OSError(f"refusing to prune run bundle outside '{runs_root}'") + current = os.stat(candidate, dir_fd=root_fd, follow_symlinks=False) + if not stat.S_ISDIR(current.st_mode): + raise OSError(f"refusing to prune non-directory run bundle '{path}'") + _remove_tree_at(root_fd, candidate) + finally: + os.close(root_fd) + + +def _remove_tree_at(parent_fd: int, name: str) -> None: + child_fd = os.open(name, _directory_open_flags(), dir_fd=parent_fd) + try: + for entry in list(os.scandir(child_fd)): + entry_stat = entry.stat(follow_symlinks=False) + if stat.S_ISDIR(entry_stat.st_mode): + _remove_tree_at(child_fd, entry.name) + else: + os.unlink(entry.name, dir_fd=child_fd) + finally: + os.close(child_fd) + os.rmdir(name, dir_fd=parent_fd) + + +def _open_directory_nofollow(path: Path) -> int: + absolute = path.resolve(strict=True) + if not hasattr(os, "O_DIRECTORY") or not hasattr(os, "O_NOFOLLOW"): + return os.open(absolute, os.O_RDONLY) + descriptor = os.open( + absolute.anchor, + _directory_open_flags(), + ) + try: + for component in absolute.relative_to(Path(absolute.anchor)).parts: + next_descriptor = os.open(component, _directory_open_flags(), dir_fd=descriptor) + os.close(descriptor) + descriptor = next_descriptor + return descriptor + except BaseException: + os.close(descriptor) + raise + + +def _safe_resolved_path(path: Path) -> Path: + try: + return Path(path).resolve(strict=False) + except OSError: + return Path(path).absolute() + + +@contextmanager +def _retention_lock(runs_root: Path) -> Iterator[None]: + lock_path = runs_root / _RUN_LOCK_NAME + lock_path.parent.mkdir(parents=True, exist_ok=True) + if lock_path.is_symlink(): + raise OSError(f"refusing to use symlinked retention lock '{lock_path}'") + if not lock_path.exists(): + try: + fd = os.open(lock_path, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600) + os.close(fd) + except FileExistsError: + pass + restrict_file(lock_path) + stream = lock_path.open("a+b") + try: + _lock_retention_stream(stream) + yield + finally: + try: + _unlock_retention_stream(stream) + finally: + stream.close() + + +def _lock_retention_stream(stream: object) -> None: + fd = stream.fileno() # type: ignore[attr-defined] + if _fcntl is not None: + _fcntl.flock(fd, _fcntl.LOCK_EX) + elif _msvcrt is not None: # pragma: no cover - Windows + stream.seek(0) # type: ignore[attr-defined] + _msvcrt.locking(fd, _msvcrt.LK_LOCK, 1) + + +def _unlock_retention_stream(stream: object) -> None: + fd = stream.fileno() # type: ignore[attr-defined] + if _fcntl is not None: + _fcntl.flock(fd, _fcntl.LOCK_UN) + elif _msvcrt is not None: # pragma: no cover - Windows + stream.seek(0) # type: ignore[attr-defined] + _msvcrt.locking(fd, _msvcrt.LK_UNLCK, 1) + + def _read_log_index(path: Path) -> set[Path] | None: try: payload = json.loads(path.read_text(encoding="utf-8")) diff --git a/lib/python/base_cli/app.py b/lib/python/base_cli/app.py index 5ca4f1e..7639b96 100644 --- a/lib/python/base_cli/app.py +++ b/lib/python/base_cli/app.py @@ -31,6 +31,7 @@ create_owned_runtime_directory, create_runtime_directory, prune_log_files, + prune_run_bundles, ) from .attachment import AttachmentContract from .config import ConfigSnapshot @@ -51,6 +52,7 @@ normalize_cli_name, ) from .profile import CliProfile +from .runtime import RetentionPolicy from .redaction import ( REDACTED, RedactionPlan, @@ -465,9 +467,38 @@ def __init__( max_log_files: int | None = None, profile: CliProfile | None = None, lifecycle_options: LifecycleOptions | None = None, + retention: RetentionPolicy | None = None, + max_run_bundles: int | None = None, + max_run_age_seconds: float | None = None, + max_run_total_bytes: int | None = None, ) -> None: if max_log_files is not None and max_log_files < 1: raise ValueError("max_log_files must be greater than 0 when set.") + if retention is not None and not isinstance(retention, RetentionPolicy): + raise TypeError("retention must be a RetentionPolicy instance or None.") + if retention is not None and any( + value is not None + for value in (max_run_bundles, max_run_age_seconds, max_run_total_bytes) + ): + raise ValueError("pass either retention or individual run retention bounds, not both.") + if retention is not None: + self.retention: RetentionPolicy | None = retention + elif any( + value is not None + for value in (max_run_bundles, max_run_age_seconds, max_run_total_bytes) + ): + self.retention = RetentionPolicy( + max_bundles=max_run_bundles, + max_age_seconds=max_run_age_seconds, + max_total_bytes=max_run_total_bytes, + ) + elif max_log_files is None: + self.retention = RetentionPolicy.safe_defaults() + else: + # Keep the original per-file option's behavior for explicitly + # opted-in legacy consumers; modern bundles are still handled by + # the compatibility path below. + self.retention = None self._registration_lock = RLock() self._registration_state = _REGISTRATION_OPEN self._name = normalize_cli_name(name or sys.argv[0]) @@ -1197,11 +1228,38 @@ def _create_context( target = f"persistent log file '{log_file}'" if log_file is not None else "stderr logging" raise RuntimeDirectoryError(f"Unable to configure {target}: {exc}") from exc context.log.debug("cli=%s run_id=%s environment=%s", self.name, run_id, environment) - retention_limit = self.max_log_files - if retention_limit is None and context.json_output: - retention_limit = _JSON_DEFAULT_MAX_LOG_FILES - if retention_limit is not None and uses_default_log_file and log_file is not None: - prune_log_files(layout.owner_root / "runs", log_file, retention_limit, context.log) + if uses_default_log_file and log_file is not None: + if self.retention is not None: + prune_run_bundles( + layout.owner_root / "runs", + layout.run_root, + policy=self.retention, + logger=context.log, + ) + elif self.max_log_files is not None: + # Compatibility for the original public option. The + # legacy pass handles pre-metadata flat log directories; + # metadata-backed runs are routed to bundle retention by + # the helper itself. + prune_log_files( + layout.owner_root / "runs", + log_file, + self.max_log_files, + context.log, + ) + prune_run_bundles( + layout.owner_root / "runs", + layout.run_root, + policy=RetentionPolicy(max_bundles=self.max_log_files), + logger=context.log, + ) + elif context.json_output: + prune_run_bundles( + layout.owner_root / "runs", + layout.run_root, + policy=RetentionPolicy(max_bundles=_JSON_DEFAULT_MAX_LOG_FILES), + logger=context.log, + ) if runtime.write_identity and selected_project_root is not None and not dry_run and self.log_to_file: try: diff --git a/lib/python/base_cli/runtime.py b/lib/python/base_cli/runtime.py index 0fda5b8..51561d7 100644 --- a/lib/python/base_cli/runtime.py +++ b/lib/python/base_cli/runtime.py @@ -5,7 +5,45 @@ from dataclasses import dataclass from pathlib import Path -__all__ = ["RuntimeLayout"] +__all__ = ["RetentionPolicy", "RuntimeLayout"] + + +DEFAULT_MAX_BUNDLES = 20 +DEFAULT_MAX_AGE_SECONDS = 30 * 24 * 60 * 60 +DEFAULT_MAX_TOTAL_BYTES = 512 * 1024 * 1024 + + +@dataclass(frozen=True) +class RetentionPolicy: + """Bounds for complete invocation bundles. + + ``None`` disables an individual bound. Protected bundles (the active + invocation, inherited parent bundles, and bundles marked ``preserve`` in + ``run.json``) are always retained, even when that means a bound cannot be + met exactly. + """ + + max_bundles: int | None = None + max_age_seconds: float | None = None + max_total_bytes: int | None = None + + @classmethod + def safe_defaults(cls) -> "RetentionPolicy": + """Return the bounded policy used by a durable :class:`App` by default.""" + + return cls( + max_bundles=DEFAULT_MAX_BUNDLES, + max_age_seconds=DEFAULT_MAX_AGE_SECONDS, + max_total_bytes=DEFAULT_MAX_TOTAL_BYTES, + ) + + def __post_init__(self) -> None: + if self.max_bundles is not None and self.max_bundles < 1: + raise ValueError("max_bundles must be greater than 0 when set.") + if self.max_age_seconds is not None and self.max_age_seconds < 0: + raise ValueError("max_age_seconds must be non-negative when set.") + if self.max_total_bytes is not None and self.max_total_bytes < 1: + raise ValueError("max_total_bytes must be greater than 0 when set.") @dataclass(frozen=True) diff --git a/tests/test_run_bundle_retention.py b/tests/test_run_bundle_retention.py new file mode 100644 index 0000000..110adba --- /dev/null +++ b/tests/test_run_bundle_retention.py @@ -0,0 +1,128 @@ +from __future__ import annotations + +import logging +import tempfile +import unittest +from pathlib import Path +from unittest import mock + +from base_cli import RetentionPolicy +from base_cli._private_files import write_private_json +from base_cli._runtime import prune_run_bundles + + +def _bundle(root: Path, name: str, *, status: str = "ok", started_at: str = "2020-01-01T00:00:00Z", size: int = 1, preserve: bool = False) -> Path: + path = root / name + (path / "logs").mkdir(parents=True) + (path / "logs" / "primary.log").write_bytes(b"x" * size) + write_private_json( + path / "run.json", + { + "run_id": name, + "status": status, + "started_at": started_at, + "preserve": preserve, + }, + ) + return path + + +class RunBundleRetentionTests(unittest.TestCase): + def test_count_removes_complete_bundles_as_units(self) -> None: + with tempfile.TemporaryDirectory() as tmpdir: + root = Path(tmpdir) / "runs" + root.mkdir() + old = _bundle(root, "old", started_at="2020-01-01T00:00:00Z") + newer = _bundle(root, "newer", started_at="2020-01-02T00:00:00Z") + (old / "tmp").mkdir() + (old / "tmp" / "diagnostic.txt").write_text("keep with bundle", encoding="utf-8") + + prune_run_bundles( + root, + root / "active", + policy=RetentionPolicy(max_bundles=2), + logger=logging.getLogger(__name__), + ) + + self.assertFalse(old.exists()) + self.assertTrue(newer.exists()) + self.assertTrue((root / ".base-cli-run-index.json").is_file()) + + def test_preserved_and_running_bundles_survive(self) -> None: + with tempfile.TemporaryDirectory() as tmpdir: + root = Path(tmpdir) / "runs" + root.mkdir() + preserved = _bundle(root, "preserved", preserve=True) + running = _bundle(root, "running", status="running") + removable = _bundle(root, "removable", started_at="2020-01-01T00:00:00Z") + + prune_run_bundles( + root, + root / "active", + policy=RetentionPolicy(max_bundles=1), + logger=logging.getLogger(__name__), + ) + + self.assertTrue(preserved.exists()) + self.assertTrue(running.exists()) + self.assertFalse(removable.exists()) + + def test_stale_running_bundle_is_recoverable_with_age_bound(self) -> None: + with tempfile.TemporaryDirectory() as tmpdir: + root = Path(tmpdir) / "runs" + root.mkdir() + stale = _bundle(root, "stale", status="running", started_at="2020-01-01T00:00:00Z") + + prune_run_bundles( + root, + root / "active", + policy=RetentionPolicy(max_age_seconds=60), + logger=logging.getLogger(__name__), + now=1_600_000_000, + ) + + self.assertFalse(stale.exists()) + + def test_symlink_bundle_is_not_followed_or_removed(self) -> None: + with tempfile.TemporaryDirectory() as tmpdir: + root = Path(tmpdir) / "runs" + external = Path(tmpdir) / "external" + root.mkdir() + external.mkdir() + victim = external / "victim.txt" + victim.write_text("do not delete", encoding="utf-8") + (root / "linked").symlink_to(external, target_is_directory=True) + + prune_run_bundles( + root, + root / "active", + policy=RetentionPolicy(max_bundles=1), + logger=logging.getLogger(__name__), + ) + + self.assertTrue(victim.exists()) + self.assertTrue((root / "linked").is_symlink()) + + def test_failed_atomic_json_write_preserves_previous_snapshot(self) -> None: + with tempfile.TemporaryDirectory() as tmpdir: + path = Path(tmpdir) / "run.json" + write_private_json(path, {"status": "ok", "run_id": "stable"}) + with mock.patch("base_cli._private_files.json.dump", side_effect=TypeError("boom")): + with self.assertRaises(TypeError): + write_private_json(path, {"status": "error"}) + self.assertEqual(path.read_text(encoding="utf-8").strip(), '{"run_id": "stable", "status": "ok"}') + + def test_atomic_json_write_refuses_symlink_destination(self) -> None: + with tempfile.TemporaryDirectory() as tmpdir: + root = Path(tmpdir) + victim = root / "victim.json" + victim.write_text("unchanged", encoding="utf-8") + destination = root / "run.json" + destination.symlink_to(victim) + with self.assertRaises(OSError): + write_private_json(destination, {"status": "error"}) + self.assertEqual(victim.read_text(encoding="utf-8"), "unchanged") + + +if __name__ == "__main__": + unittest.main()