mirror of
https://github.com/openswarm-ai/openswarm.git
synced 2026-08-20 19:52:23 +02:00
112 lines
4.2 KiB
Python
112 lines
4.2 KiB
Python
"""Workspaces on disk that no app record points at, and what reclaiming them would free.
|
|
|
|
Deleting an app used to leave its whole source tree behind (ENG-268), so installs carry orphans from
|
|
every app ever discarded: 9 of them, ~0.85GB, measured on one machine. That leak is fixed at the
|
|
delete path, but the existing pile is still there and nothing surfaces it.
|
|
|
|
Deliberately a REPORT plus an explicit delete, never an unattended sweep. `recover_orphaned_apps`
|
|
re-registers orphans that still carry a real name, so a background cleaner racing it would be two
|
|
jobs disagreeing about the same folder, and one of them silently destroying work the other is trying
|
|
to restore. Listing is free and safe; deleting is the user's call, one id at a time.
|
|
"""
|
|
|
|
import os
|
|
import shutil
|
|
from typing import List, Optional
|
|
|
|
from pydantic import BaseModel, ConfigDict
|
|
from typeguard import typechecked
|
|
|
|
from backend.config.json_store import read_json_or_none
|
|
from backend.config.paths import OUTPUTS_DIR, OUTPUTS_WORKSPACE_DIR
|
|
|
|
|
|
class OrphanWorkspace(BaseModel):
|
|
model_config = ConfigDict(validate_assignment=True)
|
|
workspace_id: str
|
|
name: str
|
|
bytes_on_disk: int
|
|
# node_modules is a symlink farm into the shared template cache, so a naive du triple-counts it
|
|
# across apps and reports a number that would not actually come back.
|
|
reclaimable_bytes: int
|
|
|
|
|
|
@typechecked
|
|
def p_tree_bytes(path: str, follow_symlinks: bool = False) -> int:
|
|
total = 0
|
|
for dirpath, dirnames, filenames in os.walk(path, followlinks=follow_symlinks):
|
|
if not follow_symlinks and os.path.basename(dirpath) == "node_modules":
|
|
dirnames[:] = []
|
|
continue
|
|
for fn in filenames:
|
|
fp = os.path.join(dirpath, fn)
|
|
try:
|
|
if not follow_symlinks and os.path.islink(fp):
|
|
continue
|
|
total += os.path.getsize(fp)
|
|
except OSError:
|
|
continue
|
|
return total
|
|
|
|
|
|
@typechecked
|
|
def p_referenced_workspace_ids() -> set:
|
|
"""Every workspace some app record still points at."""
|
|
out = set()
|
|
if not os.path.isdir(OUTPUTS_DIR):
|
|
return out
|
|
for fn in os.listdir(OUTPUTS_DIR):
|
|
if not fn.endswith(".json"):
|
|
continue
|
|
rec = read_json_or_none(os.path.join(OUTPUTS_DIR, fn))
|
|
wsid = (rec or {}).get("workspace_id")
|
|
if isinstance(wsid, str) and wsid:
|
|
out.add(wsid)
|
|
return out
|
|
|
|
|
|
@typechecked
|
|
def p_workspace_name(workspace_dir: str) -> str:
|
|
meta = read_json_or_none(os.path.join(workspace_dir, "meta.json")) or {}
|
|
name = meta.get("name")
|
|
return name.strip() if isinstance(name, str) and name.strip() else "(unnamed)"
|
|
|
|
|
|
@typechecked
|
|
def list_orphan_workspaces() -> List[OrphanWorkspace]:
|
|
"""Read-only. Never deletes, so it is safe to call from anywhere, including a status surface."""
|
|
if not os.path.isdir(OUTPUTS_WORKSPACE_DIR):
|
|
return []
|
|
referenced = p_referenced_workspace_ids()
|
|
out: List[OrphanWorkspace] = []
|
|
for wsid in sorted(os.listdir(OUTPUTS_WORKSPACE_DIR)):
|
|
path = os.path.join(OUTPUTS_WORKSPACE_DIR, wsid)
|
|
if not os.path.isdir(path) or wsid in referenced:
|
|
continue
|
|
out.append(OrphanWorkspace(
|
|
workspace_id=wsid,
|
|
name=p_workspace_name(path),
|
|
bytes_on_disk=p_tree_bytes(path, follow_symlinks=False),
|
|
reclaimable_bytes=p_tree_bytes(path, follow_symlinks=False),
|
|
))
|
|
return out
|
|
|
|
|
|
@typechecked
|
|
def delete_orphan_workspace(workspace_id: str) -> Optional[int]:
|
|
"""Remove ONE orphan by id, refusing anything still referenced or outside the root.
|
|
|
|
Returns the bytes freed, or None when the id is not a deletable orphan. Same realpath +
|
|
component-boundary guard the delete path uses: an id is stored data and rmtree is not a call to
|
|
take on trust.
|
|
"""
|
|
if workspace_id in p_referenced_workspace_ids():
|
|
return None
|
|
root = os.path.realpath(OUTPUTS_WORKSPACE_DIR)
|
|
target = os.path.realpath(os.path.join(root, workspace_id))
|
|
if target == root or not target.startswith(root + os.sep) or not os.path.isdir(target):
|
|
return None
|
|
freed = p_tree_bytes(target, follow_symlinks=False)
|
|
shutil.rmtree(target, ignore_errors=True)
|
|
return freed
|