"""TTL and idle-reap expiry candidate selection.""" from __future__ import annotations from dataclasses import dataclass from datetime import UTC, datetime from typing import Literal from sandboxer.lifecycle.store import SandboxStore from sandboxer.lifecycle.ttl import is_idle_expired, is_past_expiry from sandboxer.models import SandboxState, SandboxStatus from sandboxer.profiles.loader import load_profile ExpireReason = Literal["ttl", "idle"] @dataclass class ExpireCandidate: sandbox_id: str profile_id: str reason: ExpireReason expires_at: datetime | None = None updated_at: datetime | None = None _LIVE_STATES = frozenset({SandboxState.READY, SandboxState.ACTIVE}) def find_expire_candidates( store: SandboxStore, *, now: datetime | None = None, ) -> list[ExpireCandidate]: ref = now or datetime.now(UTC) candidates: list[ExpireCandidate] = [] seen: set[str] = set() for status in store.list_all(): if status.state not in _LIVE_STATES: continue if is_past_expiry(status.expires_at, now=ref): candidates.append( ExpireCandidate( sandbox_id=status.sandbox_id, profile_id=status.profile_id, reason="ttl", expires_at=status.expires_at, ) ) seen.add(status.sandbox_id) continue try: profile = load_profile(status.profile_id) except FileNotFoundError: continue if is_idle_expired(status.updated_at, profile.ttl.idle_reap, now=ref): candidates.append( ExpireCandidate( sandbox_id=status.sandbox_id, profile_id=status.profile_id, reason="idle", updated_at=status.updated_at, ) ) seen.add(status.sandbox_id) return sorted(candidates, key=lambda c: c.sandbox_id) def apply_expired_state(status: SandboxStatus, *, now: datetime | None = None) -> SandboxStatus: ref = now or datetime.now(UTC) status.state = SandboxState.EXPIRED status.updated_at = ref return status