refactor(utils): one unlink_files_older_than for age-based cache sweeps

Six cleanup sites hand-wrote "list a directory, unlink files whose
mtime is older than N": bot_relay had unlink_files_older_than (and
bot_mode_dm imported it from there), plus inline copies in kanban
worker-log GC, the gateway media caches, voice recordings, terminal
breadcrumbs and blocked-script cleanup.

The helper moves to utils (next to the other file helpers) and takes an
age in seconds and an optional `now`, so callers stop computing cutoffs
by hand. It never raises: a missing directory is 0, an entry that cannot
be stat'ed or unlinked is skipped for the next sweep.

Behaviour note, towards "never raises": the gateway media sweep and the
kanban worker-log GC no longer raise on a directory that vanished or
cannot be listed.

Left alone: sweeps that remove directories (delegation_live_log,
code_kernel staging), chmod read-only files first (gitlock), group
files before deciding (terminal temp cache), keep files a transcript
still references (tool_result_storage spillover), or refuse to follow
symlinks (staged uploads).

gateway/platforms/base.py

@@ -21,7 +21,7 @@ from abc import ABC, abstractmethod
 from hermes_dir_acl import make_user_dir, unusable_dir_remedy
 from gateway.platforms.base_log_safety import safe_url_for_log
 
-from utils import normalize_proxy_url
+from utils import normalize_proxy_url, unlink_files_older_than
 from agent.i18n import t
 from agent.retry_utils import jittered_backoff
 from agent.proxy_bypass import first_proxy_env_value, should_bypass_proxy as _should_bypass_proxy
@@ -543,7 +543,7 @@ def _cache_dir_accessors(kind: str, constant_name: str, new_subpath: str, old_na
         return d
 
     def cleanup(max_age_hours: int = MEDIA_CACHE_MAX_AGE_HOURS) -> int:
-        return _cleanup_cache_dir(get_dir(), max_age_hours)
+        return unlink_files_older_than(get_dir(), "*", max_age_hours * 3600)
     get_dir.__name__ = get_dir.__qualname__ = f"get_{kind}_cache_dir"
     cleanup.__name__ = cleanup.__qualname__ = f"cleanup_{kind}_cache"
     return get_dir, cleanup
@@ -669,18 +669,6 @@ async def cache_image_from_url(url: str, ext: str = ".jpg", retries: int = 2) ->
         cache_fn=cache_image_from_bytes, log_label="Media")
 
 
-def _cleanup_cache_dir(cache_dir: Path, max_age_hours: int) -> int:
-    """Delete files in *cache_dir* older than *max_age_hours*; return the count removed."""
-    cutoff = time.time() - (max_age_hours * 3600)
-    removed = 0
-    for f in cache_dir.iterdir():
-        if f.is_file() and f.stat().st_mtime < cutoff:
-            with contextlib.suppress(OSError):
-                f.unlink()
-                removed += 1
-    return removed
-
-
 # Audio cache utilities (same pattern as images; feeds the STT tool).
 AUDIO_CACHE_DIR = get_hermes_dir("cache/audio", "audio_cache")
 get_audio_cache_dir, cleanup_audio_cache = _cache_dir_accessors(

hermes_cli/kanban_db.py

@@ -31,6 +31,7 @@ from typing import Any, Iterable, Optional
 
 from hermes_cli.kanban_workflow import DEFAULT_STATUSES as VALID_STATUSES
 from toolsets import get_toolset_names
+from utils import unlink_files_older_than
 
 _log = logging.getLogger(__name__)
 
@@ -4038,14 +4039,7 @@ def gc_worker_logs(*, older_than_seconds: int = 30 * 24 * 3600, board: Optional[
     log_dir = worker_logs_dir(board=board)
     if not log_dir.exists():
         return 0
-    cutoff = time.time() - older_than_seconds
-    removed = 0
-    for p in log_dir.iterdir():
-        with contextlib.suppress(OSError):
-            if p.is_file() and p.stat().st_mtime < cutoff:
-                p.unlink()
-                removed += 1
-    return removed
+    return unlink_files_older_than(log_dir, "*", older_than_seconds)
 
 
 # --- Worker log accessor ---

hermes_cli/terminal_breadcrumbs.py

@@ -11,7 +11,7 @@ import sys
 import time
 from pathlib import Path
 from typing import Optional
-from utils import atomic_json_write
+from utils import atomic_json_write, unlink_files_older_than
 
 # Multiplexer / terminal-emulator identity env vars, checked in order when no real tty path is
 # available (e.g. stdin piped but stdout still a pty owned by a known terminal).
@@ -61,19 +61,6 @@ def is_enabled() -> bool:
         return True
 
 
-def _prune_stale(directory: Path, now: float) -> None:
-    """Best-effort removal of breadcrumbs older than the staleness window."""
-    try:
-        for entry in directory.iterdir():
-            try:
-                if entry.is_file() and now - entry.stat().st_mtime > _STALE_AFTER_SECONDS:
-                    entry.unlink()
-            except OSError:
-                continue
-    except OSError:
-        pass
-
-
 def write_breadcrumb(session_id: str, cwd: Optional[str] = None) -> None:
     """Record that this terminal's live session is ``session_id``. Never raises; no-op when the
     feature is disabled, the session id is empty, or no terminal identity exists."""
@@ -89,7 +76,7 @@ def write_breadcrumb(session_id: str, cwd: Optional[str] = None) -> None:
         now = time.time()
         payload = {"session_id": session_id, "cwd": cwd or os.getcwd(), "ts": now}
         atomic_json_write(directory / terminal_id, payload, indent=None)
-        _prune_stale(directory, now)
+        unlink_files_older_than(directory, "*", _STALE_AFTER_SECONDS, now=now)
     except Exception:
         pass
 

tests/test_utils_unlink_files_older_than.py

@@ -0,0 +1,30 @@
+"""``utils.unlink_files_older_than``: the shared age-based file sweep."""
+
+import os
+import time
+
+from utils import unlink_files_older_than
+
+
+def _aged(path, age_seconds: float, now: float):
+    path.write_text("x", encoding="utf-8")
+    os.utime(path, (now - age_seconds, now - age_seconds))
+    return path
+
+
+def test_removes_only_matching_files_older_than_the_age(tmp_path):
+    now = time.time()
+    old = _aged(tmp_path / "old.json", 120, now)
+    fresh = _aged(tmp_path / "fresh.json", 10, now)
+    other = _aged(tmp_path / "old.txt", 120, now)
+    (tmp_path / "olddir.json").mkdir()
+    os.utime(tmp_path / "olddir.json", (now - 120, now - 120))
+
+    assert unlink_files_older_than(tmp_path, "*.json", 60, now=now) == 1
+
+    assert not old.exists()
+    assert fresh.exists() and other.exists() and (tmp_path / "olddir.json").is_dir()
+
+
+def test_missing_directory_is_zero_not_an_error(tmp_path):
+    assert unlink_files_older_than(tmp_path / "absent", "*", 0) == 0

tools/approval_floors.py

@@ -7,13 +7,13 @@ allowlist runs after. Session state stays in ``tools.approval`` and is read
 through it at call time.
 """
 
-import contextlib
 import fnmatch
 import logging
 import re
 import time
 import uuid
 from tools import approval_context as _ctx
+from utils import unlink_files_older_than
 from tools.approval_detection import (
     _MALFORMED_EXEC_DESCRIPTION, _PARSER_LIMIT_DESCRIPTION, _deny_command_variants)
 
@@ -64,11 +64,7 @@ def _save_blocked_payload(command: str) -> str | None:
         script_dir = get_hermes_home() / "cache" / "blocked-scripts"
         script_dir.mkdir(parents=True, exist_ok=True)
         # Opportunistic cleanup: blocked payloads older than 7 days.
-        cutoff = time.time() - 7 * 86400
-        for old in script_dir.glob("blocked-*.sh"):
-            with contextlib.suppress(OSError):
-                if old.stat().st_mtime < cutoff:
-                    old.unlink()
+        unlink_files_older_than(script_dir, "blocked-*.sh", 7 * 86400)
         path = script_dir / f"blocked-{int(time.time())}-{uuid.uuid4().hex[:8]}.sh"
         path.write_text(
             "#!/usr/bin/env bash\n"

tools/bot_mode_dm.py

@@ -366,7 +366,6 @@ def cleanup_bot_dm_cache(max_age_hours: float = _DM_STALE_SECONDS / 3600, *, now
     """Delete orphaned DM payload files older than *max_age_hours*; returns count.
     Same contract as the other ``cleanup_*_cache`` helpers (hourly gateway housekeeping);
     legacy temp-root locations from versions predating the dedicated directory are swept too."""
-    cutoff = (time.time() if now is None else now) - max_age_hours * 3600
     temp_root = Path(tempfile.gettempdir())
     locations = [(temp_root, "hermes-dm-*.txt"), (temp_root, "hermes-relay-dm-*.txt")]
     with contextlib.suppress(OSError):
@@ -376,9 +375,9 @@ def cleanup_bot_dm_cache(max_age_hours: float = _DM_STALE_SECONDS / 3600, *, now
         # their runner on purpose — a retry replays the same delivery id from them — so the
         # orphans of runners that never settled are swept here too.
         locations.append((dm_dir, "*.live.json"))
-    from tools.bot_relay import unlink_files_older_than
+    from utils import unlink_files_older_than
 
-    return sum(unlink_files_older_than(d, pattern, cutoff) for d, pattern in locations)
+    return sum(unlink_files_older_than(d, pattern, max_age_hours * 3600, now=now) for d, pattern in locations)
 
 
 def _unlink_dm_file(path: str) -> None:

tools/bot_relay.py

@@ -27,7 +27,7 @@ from pathlib import Path
 from typing import Any, Iterator, Mapping, Optional
 
 from tools.bot_mode_probe import _default_home, _hermes_root, alias_forms
-from utils import atomic_json_write
+from utils import atomic_json_write, unlink_files_older_than
 
 logger = logging.getLogger(__name__)
 
@@ -440,21 +440,9 @@ def write_reply(root: Path | str, envelope_id: str, *, reply: str = "", error: s
     return path
 
 
-def unlink_files_older_than(directory: Path, pattern: str, cutoff: float) -> int:
-    """Unlink regular files matching ``pattern`` with mtime before ``cutoff``; returns count. Never raises."""
-    removed = 0
-    with contextlib.suppress(OSError):
-        for path in directory.glob(pattern):
-            with contextlib.suppress(OSError):
-                if path.is_file() and path.stat().st_mtime < cutoff:
-                    path.unlink()
-                    removed += 1
-    return removed
-
-
 def _sweep_stale(base: Path, *, now: float | None = None) -> int:
-    cutoff = (time.time() if now is None else now) - STALE_AFTER_SECONDS
-    return sum(unlink_files_older_than(base / sub, "*.json", cutoff) for sub in (CLAIMED_DIR, REPLIES_DIR, OUTBOX_DIR))
+    return sum(unlink_files_older_than(base / sub, "*.json", STALE_AFTER_SECONDS, now=now)
+               for sub in (CLAIMED_DIR, REPLIES_DIR, OUTBOX_DIR))
 
 
 def cleanup_bot_relay_artifacts(max_age_hours: float | None = None) -> int:

tools/voice_mode.py

@@ -1541,15 +1541,9 @@ def check_voice_requirements() -> dict[str, Any]:
 # ── Temp file cleanup ──
 def cleanup_temp_recordings(max_age_seconds: int = 3600) -> int:
     """Remove ``recording_*.wav`` temp files older than *max_age_seconds*; returns the count."""
-    if not os.path.isdir(_TEMP_DIR):
-        return 0
-    deleted, now = 0, time.time()
-    for entry in os.scandir(_TEMP_DIR):
-        if entry.is_file() and entry.name.startswith("recording_") and entry.name.endswith(".wav"):
-            with suppress(OSError):
-                if now - entry.stat().st_mtime > max_age_seconds:
-                    os.unlink(entry.path)
-                    deleted += 1
+    from utils import unlink_files_older_than
+
+    deleted = unlink_files_older_than(_TEMP_DIR, "recording_*.wav", max_age_seconds)
     if deleted:
         logger.debug("Cleaned up %d old voice recordings", deleted)
     return deleted

utils.py

@@ -263,6 +263,25 @@ def fsync_directory(path: str | Path) -> None:
         os.close(fd)
 
 
+def unlink_files_older_than(directory: str | Path, pattern: str, max_age_seconds: float, *,
+                            now: float | None = None) -> int:
+    """Delete regular files in *directory* matching *pattern* whose mtime is more than
+    *max_age_seconds* before *now* (default: the current time); return how many were removed.
+
+    The one age-based cache sweep. Never raises: a missing directory, or an entry that cannot be
+    stat'ed or unlinked, is skipped (the next sweep retries it).
+    """
+    cutoff = (time.time() if now is None else now) - max_age_seconds
+    removed = 0
+    with suppress(OSError):
+        for path in Path(directory).glob(pattern):
+            with suppress(OSError):
+                if path.is_file() and path.stat().st_mtime < cutoff:
+                    path.unlink()
+                    removed += 1
+    return removed
+
+
 def rmtree_readonly(path: str | Path, *, ignore_errors: bool = False) -> None:
     """``shutil.rmtree`` that can also delete read-only trees.