the-nexus/mempalace/retain_closets.py

#!/usr/bin/env python3
"""
retain_closets.py — Retention policy enforcement for fleet palace closets.

Removes closet files older than a configurable retention window (default: 90 days).
Run this on the Alpha host (or any fleet palace directory) to enforce the
closet aging policy described in #1083.

Usage:
    # Dry-run: show what would be removed (no deletions)
    python mempalace/retain_closets.py --dry-run

    # Enforce 90-day retention (default)
    python mempalace/retain_closets.py

    # Custom retention window
    python mempalace/retain_closets.py --days 30

    # Custom palace path
    python mempalace/retain_closets.py /data/fleet --days 90

Exits:
    0 — success (clean, or pruned without error)
    1 — error (e.g., palace directory not found)

Refs: #1083, #1075
"""

from __future__ import annotations

import argparse
import os
import sys
import time
from dataclasses import dataclass, field
from pathlib import Path

DEFAULT_RETENTION_DAYS = 90
DEFAULT_PALACE_PATH = "/var/lib/mempalace/fleet"


@dataclass
class RetentionResult:
    scanned: int = 0
    removed: int = 0
    kept: int = 0
    errors: list[str] = field(default_factory=list)

    @property
    def ok(self) -> bool:
        return len(self.errors) == 0


def _file_age_days(path: Path) -> float:
    """Return the age of a file in days based on mtime."""
    mtime = path.stat().st_mtime
    now = time.time()
    return (now - mtime) / 86400.0


def enforce_retention(
    palace_dir: Path,
    retention_days: int = DEFAULT_RETENTION_DAYS,
    dry_run: bool = False,
) -> RetentionResult:
    """
    Remove *.closet.json files older than *retention_days* from *palace_dir*.

    Only closet files are pruned — raw drawer files are never present in a
    compliant fleet palace, so this script does not touch them.

    Args:
        palace_dir: Root directory of the fleet palace to scan.
        retention_days: Files older than this many days will be removed.
        dry_run: If True, report what would be removed but make no changes.

    Returns:
        RetentionResult with counts and any errors.
    """
    result = RetentionResult()

    for closet_file in sorted(palace_dir.rglob("*.closet.json")):
        result.scanned += 1
        try:
            age = _file_age_days(closet_file)
        except OSError as exc:
            result.errors.append(f"Could not stat {closet_file}: {exc}")
            continue

        if age > retention_days:
            if dry_run:
                print(
                    f"[retain_closets] DRY-RUN would remove ({age:.0f}d old): {closet_file}"
                )
                result.removed += 1
            else:
                try:
                    closet_file.unlink()
                    print(f"[retain_closets] Removed ({age:.0f}d old): {closet_file}")
                    result.removed += 1
                except OSError as exc:
                    result.errors.append(f"Could not remove {closet_file}: {exc}")
        else:
            result.kept += 1

    return result


def main(argv: list[str] | None = None) -> int:
    parser = argparse.ArgumentParser(
        description="Enforce retention policy on fleet palace closets."
    )
    parser.add_argument(
        "palace_dir",
        nargs="?",
        default=os.environ.get("FLEET_PALACE_PATH", DEFAULT_PALACE_PATH),
        help=f"Fleet palace directory (default: {DEFAULT_PALACE_PATH})",
    )
    parser.add_argument(
        "--days",
        type=int,
        default=DEFAULT_RETENTION_DAYS,
        metavar="N",
        help=f"Retention window in days (default: {DEFAULT_RETENTION_DAYS})",
    )
    parser.add_argument(
        "--dry-run",
        action="store_true",
        help="Show what would be removed without deleting anything.",
    )
    args = parser.parse_args(argv)

    palace_dir = Path(args.palace_dir)
    if not palace_dir.exists():
        print(
            f"[retain_closets] ERROR: palace directory not found: {palace_dir}",
            file=sys.stderr,
        )
        return 1

    mode = "DRY-RUN" if args.dry_run else "LIVE"
    print(
        f"[retain_closets] {mode} — scanning {palace_dir} "
        f"(retention: {args.days} days)"
    )

    result = enforce_retention(palace_dir, retention_days=args.days, dry_run=args.dry_run)

    if result.errors:
        for err in result.errors:
            print(f"[retain_closets] ERROR: {err}", file=sys.stderr)
        return 1

    action = "would remove" if args.dry_run else "removed"
    print(
        f"[retain_closets] Done — scanned {result.scanned}, "
        f"{action} {result.removed}, kept {result.kept}."
    )
    return 0


if __name__ == "__main__":
    sys.exit(main())