"""Server-side cleanup for durable session deletion results.""" from __future__ import annotations import hashlib import os import re import shutil import stat from collections.abc import Callable, Iterable, Mapping from pathlib import Path from typing import Any, Protocol from openai4s.host.data import kernel_artifact_input_dir from openai4s.storage.snapshots import WorkspaceCAS from openai4s.tools.dynamic_scopes import DynamicScopeStore class SessionDeletionStore(Protocol): def get_frame(self, frame_id: str) -> dict | None: ... def project_session_ids(self, project_id: str) -> list[str]: ... def delete_frame(self, frame_id: str) -> dict[str, Any]: ... def delete_project(self, project_id: str) -> dict[str, Any]: ... def retained_workspace_tree_ids(self) -> tuple[str, ...]: ... class SessionDeletionService: """Stop live resources, delete durable rows, then clean owned files.""" def __init__( self, store: SessionDeletionStore, *, data_dir: str | Path, cas: WorkspaceCAS, drop_runtime: Callable[[str, str], Any], drop_resume_window: Callable[[str], Any], revoke_shares: Callable[[str], Any] | None = None, release_compute: Callable[[str], Any] | None = None, cleanup_frameless_uploads: bool = False, ) -> None: self.store = store self.data_dir = Path(data_dir).expanduser().resolve() self.workspace_root = self.data_dir / "agent-workspaces" self.cas = cas self.dynamic_scopes = DynamicScopeStore( self.data_dir / "dynamic-tools" / "_scoped" ) self._drop_runtime = drop_runtime self._drop_resume_window = drop_resume_window self._revoke_shares = revoke_shares or (lambda _root_frame_id: None) # A deleted session's cluster resource is not deleted with it. The # workload is a durable row with a job behind it, and dropping the # local runtime says nothing to the scheduler -- so without this the # session vanishes from the UI while the job keeps its node, its GPUs # and its lease, with nothing left in the product that names it. # A distinct collaborator rather than extra code inside # `drop_runtime`, because folding it in there is this repo's # recurring "one guard, one of several call sites" defect. self._release_compute = release_compute or (lambda _root_frame_id: None) # The Gateway enables this only together with its always-on global # frameless-mutation/deletion barrier. Direct compositions that cannot # prove that admission boundary retain the safe "leave for sweeper" # default instead of unlinking a shared uploads basename concurrently. self.cleanup_frameless_uploads = bool(cleanup_frameless_uploads) def delete_session( self, root_frame_id: str, *, reason: str = "frame_deleted" ) -> dict[str, Any]: frame = self.store.get_frame(root_frame_id) if frame is not None: canonical = str(frame.get("root_frame_id") or frame.get("frame_id")) if canonical != root_frame_id: raise ValueError("session deletion requires a root frame id") self._drop_runtime(root_frame_id, reason) self._release_compute_safe(root_frame_id) self._revoke_shares_safe(root_frame_id) result = self.store.delete_frame(root_frame_id) cleanup = self._cleanup(result) self._drop_resume_window(root_frame_id) return {"ok": True, **cleanup} def delete_project( self, project_id: str, *, reason: str = "project_deleted" ) -> dict[str, Any]: roots = self.store.project_session_ids(project_id) for root_frame_id in roots: self._drop_runtime(root_frame_id, reason) self._release_compute_safe(root_frame_id) self._revoke_shares_safe(root_frame_id) result = self.store.delete_project(project_id) deleted_roots = tuple( dict.fromkeys( str(value) for value in result.get("root_frame_ids", ()) if value ) ) # Admission is closed by SessionRunner while this service runs. This # second pass is a fail-safe for legacy/direct Store writers that may # have inserted a root after the initial enumeration. for root_frame_id in deleted_roots: if root_frame_id not in roots: self._drop_runtime(root_frame_id, reason) self._release_compute_safe(root_frame_id) self._revoke_shares_safe(root_frame_id) cleanup = self._cleanup( result, include_uploads=self.cleanup_frameless_uploads, ) dynamic = self.dynamic_scopes.delete_project_scope(project_id) for root_frame_id in deleted_roots: self._drop_resume_window(root_frame_id) return { "ok": True, **cleanup, "freed_dynamic_events": dynamic["events"], "freed_dynamic_manifests": dynamic["manifests"], } def _release_compute_safe(self, root_frame_id: str) -> None: """Ask for the cluster resource back. Never fails the deletion. Swallowed like share revocation: a scheduler that cannot be reached must not leave a user unable to delete their own session. What it must also not do is leave *no* record, so the reconciler's durable stop request is the mechanism -- the desire survives a restart and the barrier runs on the next tick. """ try: self._release_compute(root_frame_id) except Exception: # noqa: BLE001 — deletion proceeds regardless pass def _revoke_shares_safe(self, root_frame_id: str) -> None: try: self._revoke_shares(root_frame_id) except Exception: # noqa: BLE001 - deletion must proceed regardless pass def _cleanup( self, result: Mapping[str, Any], *, include_uploads: bool = False, ) -> dict[str, Any]: roots = tuple( dict.fromkeys( str(value) for value in result.get("root_frame_ids", ()) if value ) ) # Immutable snapshots are unlinked one by one. Stage 1 also reclaims # direct frameless uploads while its global mutation barrier is held. # Live session files are reclaimed by the confined tree removal below. # ``uploads`` basenames can be reused across project scopes, so deletion # relies on the retained path/inode identities below and only removes a # direct child after no surviving version references it. trusted_roots = [self.data_dir / "artifact-versions"] if include_uploads: trusted_roots.append(self.data_dir / "uploads") freed_files = 0 skipped_files = 0 retained_paths, retained_files = self._retained_path_identities( result.get("retained_paths", ()) ) for raw_path in result.get("stale_paths", ()): if self._unlink_owned_file( raw_path, trusted_roots, retained_paths=retained_paths, retained_files=retained_files, ): freed_files += 1 else: skipped_files += 1 freed_workspaces = 0 for root_frame_id in roots: if self._remove_root_workspace(root_frame_id): freed_workspaces += 1 if self._remove_branch_workspaces(root_frame_id): freed_workspaces += 1 if self._remove_dynamic_tools(root_frame_id): freed_workspaces += 1 if self._remove_session_import(root_frame_id): freed_workspaces += 1 if self._remove_kernel_artifact_inputs(root_frame_id): freed_workspaces += 1 cas = self.cas.release_trees( result.get("cas_tree_ids", ()), retained_tree_ids=result.get("retained_cas_tree_ids", ()), retained_tree_ids_provider=self.store.retained_workspace_tree_ids, ) return { "deleted": bool(result.get("deleted")), "freed_sessions": len(roots), "freed_files": freed_files, "skipped_unowned_files": skipped_files, "freed_workspaces": freed_workspaces, "freed_cas_trees": cas["trees"], "freed_cas_blobs": cas["blobs"], "shared_cas_trees": cas["shared_trees"], "deleted_rows": dict(result.get("deleted_rows") or {}), } @staticmethod def _unlink_owned_file( raw_path: Any, trusted_roots: Iterable[Path], *, retained_paths: set[str], retained_files: set[tuple[int, int]], ) -> bool: if not isinstance(raw_path, str) or not raw_path: return False candidate = Path(os.path.abspath(Path(raw_path).expanduser())) if not candidate.is_absolute() or candidate.name in {"", ".", ".."}: return False for raw_root in trusted_roots: lexical_root = Path(os.path.abspath(Path(raw_root).expanduser())) # Authorization is lexical and direct-child-only. Resolving a # symlinked trusted root would bless its external target, which is # the opposite of confinement. if candidate.parent != lexical_root: continue try: root_before = os.lstat(lexical_root) except OSError: continue if not stat.S_ISDIR(root_before.st_mode) or stat.S_ISLNK( root_before.st_mode ): continue supports_dir_fd = ( os.stat in os.supports_dir_fd and os.unlink in os.supports_dir_fd and os.stat in os.supports_follow_symlinks and bool(getattr(os, "O_NOFOLLOW", 0)) ) if supports_dir_fd: descriptor: int | None = None try: flags = os.O_RDONLY | getattr(os, "O_DIRECTORY", 0) flags |= getattr(os, "O_NOFOLLOW", 0) descriptor = os.open(lexical_root, flags) opened_root = os.fstat(descriptor) if not stat.S_ISDIR(opened_root.st_mode) or ( int(opened_root.st_dev), int(opened_root.st_ino), ) != (int(root_before.st_dev), int(root_before.st_ino)): return False entry = os.stat( candidate.name, dir_fd=descriptor, follow_symlinks=False, ) if not stat.S_ISREG(entry.st_mode): return False identity = (int(entry.st_dev), int(entry.st_ino)) try: resolved_key = os.path.normcase( str(candidate.resolve(strict=False)) ) except OSError: return False if resolved_key in retained_paths or identity in retained_files: return False # Re-read the directory entry immediately before unlink. # A regular->symlink swap is refused rather than deleting # even the link; a regular replacement is likewise not the # file whose identity the repository declared stale. latest = os.stat( candidate.name, dir_fd=descriptor, follow_symlinks=False, ) if ( not stat.S_ISREG(latest.st_mode) or (int(latest.st_dev), int(latest.st_ino)) != identity ): return False os.unlink(candidate.name, dir_fd=descriptor) return True except (FileNotFoundError, NotADirectoryError, OSError): return False finally: if descriptor is not None: os.close(descriptor) # Without dir-fd + no-follow there is no atomic way to keep a # checked parent bound through unlink. Leaving stale bytes for a # later supported-platform sweep is safer than a path-based delete # whose root can be swapped after its final check. return False return False @staticmethod def _retained_path_identities( paths: Iterable[Any], ) -> tuple[set[str], set[tuple[int, int]]]: resolved_paths: set[str] = set() file_ids: set[tuple[int, int]] = set() for raw_path in paths: if not isinstance(raw_path, str) or not raw_path: continue try: path = Path(raw_path).expanduser().resolve(strict=False) except OSError: continue resolved_paths.add(os.path.normcase(str(path))) identity = SessionDeletionService._file_identity(path) if identity is not None: file_ids.add(identity) return resolved_paths, file_ids @staticmethod def _file_identity(path: Path) -> tuple[int, int] | None: try: info = path.stat() except OSError: return None return int(info.st_dev), int(info.st_ino) def _remove_root_workspace(self, root_frame_id: str) -> bool: if Path(root_frame_id).name != root_frame_id or root_frame_id in {".", ".."}: return False return self._remove_owned_tree( self.workspace_root / root_frame_id, direct_parent=self.workspace_root, ) def _remove_branch_workspaces(self, root_frame_id: str) -> bool: root_key = hashlib.sha256(root_frame_id.encode("utf-8")).hexdigest()[:24] parent = self.workspace_root / ".branches" removed = self._remove_owned_tree(parent / root_key, direct_parent=parent) try: parent.rmdir() except OSError: pass return removed def _remove_dynamic_tools(self, root_frame_id: str) -> bool: safe_session = re.sub(r"[^A-Za-z0-9._-]+", "_", root_frame_id) # Avoid sanitizer collisions and the shared project/global audit tree. if safe_session != root_frame_id or safe_session == "_scoped": return False parent = self.data_dir / "dynamic-tools" return self._remove_owned_tree( parent / safe_session, direct_parent=parent, ) def _remove_session_import(self, root_frame_id: str) -> bool: if Path(root_frame_id).name != root_frame_id or root_frame_id in {".", ".."}: return False parent = self.data_dir / "session-imports" removed = self._remove_owned_tree( parent / root_frame_id, direct_parent=parent, ) try: parent.rmdir() except OSError: pass return removed def _remove_kernel_artifact_inputs(self, root_frame_id: str) -> bool: """Reclaim the exact-version copies retained for a persistent kernel.""" try: path = kernel_artifact_input_dir(self.data_dir, root_frame_id) except ValueError: return False parent = path.parent removed = self._remove_owned_tree(path, direct_parent=parent) try: parent.rmdir() except OSError: pass return removed @staticmethod def _remove_owned_tree(path: Path, *, direct_parent: Path) -> bool: try: if path.parent.resolve() != direct_parent.resolve(): return False if path.is_symlink(): path.unlink() return True if not path.exists(): return False if not path.is_dir(): return False shutil.rmtree(path) return True except OSError: return False __all__ = ["SessionDeletionService", "SessionDeletionStore"]