import base64 import calendar import json import os import re import shutil import subprocess import sys import time import urllib.error import urllib.parse import urllib.request CONFIG = json.loads(base64.b64decode("__UNIDESK_GC_REMOTE_CONFIG_BASE64__").decode("utf-8")) PROVIDER_ID = str(CONFIG.get("providerId") or "") ACTION = str(CONFIG.get("action") or "plan") OPTIONS = CONFIG.get("options") or {} REMOTE_TARGET = CONFIG.get("remoteTarget") if isinstance(CONFIG.get("remoteTarget"), dict) else {} MEMORY_CONFIG = REMOTE_TARGET.get("memoryPressure") if isinstance(REMOTE_TARGET.get("memoryPressure"), dict) else {} PVC_CONFIG = REMOTE_TARGET.get("pvcAttribution") if isinstance(REMOTE_TARGET.get("pvcAttribution"), dict) else {} CONTAINERD_CONFIG = REMOTE_TARGET.get("containerdImageCache") if isinstance(REMOTE_TARGET.get("containerdImageCache"), dict) else {} HOST_CONTAINERD_CONFIG = REMOTE_TARGET.get("hostContainerdCache") if isinstance(REMOTE_TARGET.get("hostContainerdCache"), dict) else {} LOCAL_PATH_CONFIG = REMOTE_TARGET.get("localPathStorage") if isinstance(REMOTE_TARGET.get("localPathStorage"), dict) else {} POLICY_TIMER_CONFIG = REMOTE_TARGET.get("policyTimer") if isinstance(REMOTE_TARGET.get("policyTimer"), dict) else {} HOST_DOCKER_GC_CONFIG = REMOTE_TARGET.get("hostDockerGc") if isinstance(REMOTE_TARGET.get("hostDockerGc"), dict) else {} POLICY_RUNNER_SOURCE = base64.b64decode("__UNIDESK_GC_REMOTE_POLICY_RUNNER_BASE64__").decode("utf-8") TMP_PREFIX_ALLOWLIST = [ "hwlab-agent-", "hwlab-cd-", "hwlab-cli-cicd-", "hwlab-codeagent-trace", "hwlab-desired-state-", "hwlab-g14-", "hwlab-main-", "hwlab-merge-", "hwlab-pr", "hwlab-refresh-", "hwlab-remote-", "hwlab-ts-check", "hwlab-bun-runtime-check-", "hwlab-v02-", "playwright-artifacts-", "playwright_chromiumdev_profile-", "unidesk-apply-patch-v2-perf-", "unidesk-clean-", "unidesk-code-queue", "unidesk-hwlab-cd-", "unidesk-pr", "unidesk-tran-runner", "bunx-", "codex-app-schema", "codex-app-ts", "marked-", "node-compile-cache", ] TMP_EXACT_PROTECT = set([ "/tmp/codex-apply-patch", "/tmp/codex-ipc", "/tmp/tmux-0", "/tmp/snap-private-tmp", ]) CORE_DUMP_DIR_ALLOWLIST = set([ "/root/unidesk", ]) TOOL_CACHE_ALLOWLIST = [ { "id": "npm-cacache", "path": "/root/.npm/_cacache", "description": "Delete npm content-addressable package cache; npm can rebuild it.", }, { "id": "npm-npx", "path": "/root/.npm/_npx", "description": "Delete npx package execution cache; npx can rebuild it.", }, { "id": "bun-install-cache", "path": "/root/.bun/install/cache", "description": "Delete Bun install package cache; bun can rebuild it.", }, ] REGISTRY_REPOSITORY_ROOT = "/var/lib/hwlab/registry/docker/registry/v2/repositories" REGISTRY_ROOT = "/var/lib/hwlab/registry" REGISTRY_PROTECTED_TAGS = set([ "latest", "16-alpine", "20-bookworm-slim", "node22-alpine-v1", "node22-alpine-bun-v1", "sidecar", "1b99888d3dae", ]) EXPECTED_G14_NODE = "ubuntu-rog-zephyrus-g14-ga401iv-ga401iv" REMOTE_GC_JOB_DIR = "/tmp/unidesk-gc-remote/jobs" REMOTE_GROWTH_SNAPSHOT_DIR = "/tmp/unidesk-gc-remote/growth-snapshots" REMOTE_STDOUT_JSON_LIMIT = 256 * 1024 def config_list(cfg, key, default=None): value = cfg.get(key) if isinstance(cfg, dict) else None if isinstance(value, list): return [str(item) for item in value if isinstance(item, (str, int, float)) and str(item)] return list(default or []) def config_bool(cfg, key, default=False): value = cfg.get(key) if isinstance(cfg, dict) else None if isinstance(value, bool): return value return bool(default) def config_int(cfg, key, default=0, minimum=None, maximum=None): value = cfg.get(key) if isinstance(cfg, dict) else None try: parsed = int(value) except Exception: parsed = int(default) if minimum is not None: parsed = max(int(minimum), parsed) if maximum is not None: parsed = min(int(maximum), parsed) return parsed def config_float(cfg, key, default=0.0, minimum=None, maximum=None): value = cfg.get(key) if isinstance(cfg, dict) else None try: parsed = float(value) except Exception: parsed = float(default) if minimum is not None: parsed = max(float(minimum), parsed) if maximum is not None: parsed = min(float(maximum), parsed) return parsed def config_str(cfg, key, default=""): value = cfg.get(key) if isinstance(cfg, dict) else None if isinstance(value, str) and value: return value return str(default) def host_docker_gc_section(name): value = HOST_DOCKER_GC_CONFIG.get(name) if isinstance(HOST_DOCKER_GC_CONFIG, dict) else None return value if isinstance(value, dict) else {} def host_docker_gc_enabled(name, default_enabled=True): section = host_docker_gc_section(name) if "enabled" in section: return bool(section.get("enabled")) if PROVIDER_ID.upper() == "JD01": return False return bool(default_enabled) def host_docker_gc_reason(name, fallback): reason = host_docker_gc_section(name).get("reason") return str(reason) if isinstance(reason, str) and reason else fallback def parse_size_value(value, default=None): if isinstance(value, (int, float)) and value > 0: return int(value) if not isinstance(value, str): return default match = re.match(r"^\s*([0-9]+(?:\.[0-9]+)?)\s*(b|k|kb|kib|m|mb|mib|g|gb|gib)?\s*$", value, re.I) if not match: return default unit = (match.group(2) or "b").lower() mult = 1024**3 if unit in set(["g", "gb", "gib"]) else 1024**2 if unit in set(["m", "mb", "mib"]) else 1024 if unit in set(["k", "kb", "kib"]) else 1 return int(float(match.group(1)) * mult) def now_iso(): return time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()) def command(cmd, timeout=10): try: p = subprocess.run(cmd, text=True, capture_output=True, timeout=timeout) return {"exitCode": p.returncode, "stdout": p.stdout, "stderr": p.stderr, "timedOut": False} except subprocess.TimeoutExpired as exc: return { "exitCode": None, "stdout": exc.stdout or "", "stderr": exc.stderr or ("timed out after %ss" % timeout), "timedOut": True, } except Exception as exc: return {"exitCode": None, "stdout": "", "stderr": str(exc), "timedOut": False} def bounded(result): return { "exitCode": result.get("exitCode"), "timedOut": bool(result.get("timedOut")), "stdoutTail": str(result.get("stdout") or "")[-2000:], "stderrTail": str(result.get("stderr") or "")[-2000:], } def job_id_or_none(): raw = str(OPTIONS.get("jobId") or "") if raw and re.match(r"^[A-Za-z0-9._-]{1,128}$", raw): return raw return None def job_paths(job_id): os.makedirs(REMOTE_GC_JOB_DIR, exist_ok=True) return { "state": os.path.join(REMOTE_GC_JOB_DIR, "%s.json" % job_id), "log": os.path.join(REMOTE_GC_JOB_DIR, "%s.log" % job_id), } def status_command(job_id): return "bun scripts/cli.ts gc remote %s status --job-id %s" % (PROVIDER_ID, job_id) def write_json_atomic(path, payload): tmp = "%s.tmp.%s" % (path, os.getpid()) with open(tmp, "w", encoding="utf-8") as handle: json.dump(payload, handle, ensure_ascii=False, indent=2) handle.write("\n") os.replace(tmp, path) def read_file_tail(path, limit=12000): try: size = os.path.getsize(path) with open(path, "rb") as handle: if size > limit: handle.seek(size - limit) data = handle.read() return data.decode("utf-8", errors="replace") except OSError: return "" def stdout_page(items): if not isinstance(items, list): return items raw_limit = OPTIONS.get("resultLimit") or OPTIONS.get("limit") or 50 try: limit = int(raw_limit) except Exception: limit = 50 limit = max(1, min(limit, 100)) return items[:limit] def compact_payload_for_stdout(payload, full_size_bytes, job_id=None, paths=None): compact = { "ok": payload.get("ok", True), "action": payload.get("action") or "gc remote", "providerId": payload.get("providerId") or PROVIDER_ID, "output": { "truncated": True, "reason": "stdout-size-guard", "fullResultBytes": full_size_bytes, }, } for key in [ "dryRun", "mutation", "observedAt", "status", "kind", "mode", "startedAt", "finishedAt", "error", "message", "options", "diskBefore", "diskAfter", "clusterPreflight", "clusterAfter", "summary", "policy", ]: if key in payload: compact[key] = payload[key] if job_id: state_path = paths["state"] if paths else payload.get("statePath") compact["jobId"] = job_id compact["statePath"] = state_path compact["statusCommand"] = status_command(job_id) compact["fullResult"] = { "jobId": job_id, "statePath": state_path, "statusCommand": status_command(job_id), } compact["output"]["fullResultJobId"] = job_id if "results" in payload: results = payload.get("results") or [] compact["results"] = stdout_page(results) compact["returnedResultCount"] = len(compact["results"]) compact["omittedResultCount"] = max(0, len(results) - len(compact["results"])) if isinstance(results, list) else 0 if "candidates" in payload: candidates = payload.get("candidates") or [] compact["candidates"] = stdout_page(candidates) compact["returnedCandidateCount"] = len(compact["candidates"]) compact["omittedCandidateCount"] = max(0, len(candidates) - len(compact["candidates"])) if isinstance(candidates, list) else 0 if "protected" in payload: compact["protected"] = payload["protected"] if "logTail" in payload: compact["logTail"] = str(payload.get("logTail") or "")[-12000:] return compact def emit_json(payload, persist_large=True): raw = json.dumps(payload, ensure_ascii=False, indent=2) full_size = len(raw.encode("utf-8")) if full_size <= REMOTE_STDOUT_JSON_LIMIT: print(raw) return job_id = str(payload.get("jobId") or "") paths = None if persist_large: if not job_id: provider_slug = re.sub(r"[^A-Za-z0-9._-]+", "-", PROVIDER_ID.lower()).strip("-") or "provider" job_id = "%s-gc-output-%s-%s" % (provider_slug, int(time.time()), os.getpid()) paths = job_paths(job_id) payload = dict(payload) payload.update({ "jobId": job_id, "statePath": paths["state"], "statusCommand": status_command(job_id), "outputPersistedAt": now_iso(), }) write_json_atomic(paths["state"], payload) elif job_id: paths = job_paths(job_id) compact = compact_payload_for_stdout(payload, full_size, job_id or None, paths) print(json.dumps(compact, ensure_ascii=False, indent=2)) def remote_gc_job_status(): job_id = job_id_or_none() if not job_id: return remote_gc_live_status(now_iso(), cluster_preflight()) paths = job_paths(job_id) if not os.path.isfile(paths["state"]): return { "ok": False, "error": "gc-remote-job-not-found", "jobId": job_id, "statePath": paths["state"], "logTail": read_file_tail(paths["log"]), } try: with open(paths["state"], "r", encoding="utf-8") as handle: payload = json.load(handle) except Exception as exc: return { "ok": False, "error": "gc-remote-job-state-invalid", "jobId": job_id, "message": str(exc), "statePath": paths["state"], "logTail": read_file_tail(paths["log"]), } payload["logTail"] = read_file_tail(paths["log"]) return payload def remote_gc_live_status(observed_at, preflight): memory_pressure = collect_memory_pressure() ci_storage = ci_storage_snapshot() compact_pvc = compact_pvc_attribution(ci_storage) policy = growth_watermark_policy(df_snapshot() or {}) return { "ok": True, "action": "gc remote status", "providerId": PROVIDER_ID, "dryRun": True, "mutation": False, "observedAt": observed_at, "disk": df_snapshot(), "clusterPreflight": preflight if bool(OPTIONS.get("full")) else {key: preflight.get(key) for key in ["ok", "reason", "providerId", "hostname", "expectedNode", "nodes"]}, "memoryPressure": compact_memory_pressure(memory_pressure), "pvcAttribution": compact_pvc, "policy": policy if bool(OPTIONS.get("full")) else {key: policy.get(key) for key in ["state", "recommendedAction"]}, "next": { "snapshot": "bun scripts/cli.ts gc remote %s snapshot --history-limit %s" % (PROVIDER_ID, int(OPTIONS.get("historyLimit") or 12)), "plan": "bun scripts/cli.ts gc remote %s plan --target-use-percent --limit %s" % (PROVIDER_ID, int(OPTIONS.get("limit") or 50)), "policy": "bun scripts/cli.ts gc remote %s policy plan" % PROVIDER_ID, "jobStatus": "bun scripts/cli.ts gc remote %s status --job-id " % PROVIDER_ID, }, } def path_size(path): try: if os.path.islink(path) or os.path.isfile(path): return os.lstat(path).st_size if not os.path.isdir(path): return 0 total = 0 for root, dirs, files in os.walk(path): for name in files: child = os.path.join(root, name) try: total += os.lstat(child).st_size except OSError: pass for name in dirs: child = os.path.join(root, name) try: if os.path.islink(child): total += os.lstat(child).st_size except OSError: pass return total except OSError: return 0 def du_size(path, timeout=20, fallback_walk=True): if not os.path.exists(path): return None result = command(["du", "-sxB1", path], timeout) if result["exitCode"] != 0: if not fallback_walk: return None return path_size(path) text = result["stdout"].strip() if not text: return 0 try: return int(text.split()[0]) except Exception: if not fallback_walk: return None return path_size(path) def safe_int(value, default=0): try: if value is None: return default return int(value) except Exception: return default def iso_to_epoch(value): try: return calendar.timegm(time.strptime(str(value), "%Y-%m-%dT%H:%M:%SZ")) except Exception: return None def growth_snapshot_path(): os.makedirs(REMOTE_GROWTH_SNAPSHOT_DIR, exist_ok=True) provider_slug = re.sub(r"[^A-Za-z0-9._-]+", "-", PROVIDER_ID.lower()).strip("-") or "provider" return os.path.join(REMOTE_GROWTH_SNAPSHOT_DIR, "%s.jsonl" % provider_slug) def read_growth_snapshots(limit=None): path = growth_snapshot_path() if not os.path.isfile(path): return [] try: with open(path, "r", encoding="utf-8") as handle: lines = handle.readlines() except OSError: return [] rows = [] for line in lines[-max(1, int(limit or 200)):]: line = line.strip() if not line: continue try: item = json.loads(line) except Exception: continue if isinstance(item, dict): rows.append(item) return rows def append_growth_snapshot(snapshot): path = growth_snapshot_path() os.makedirs(os.path.dirname(path), exist_ok=True) with open(path, "a", encoding="utf-8") as handle: handle.write(json.dumps(snapshot, ensure_ascii=False, sort_keys=True)) handle.write("\n") return path def source_size_item(source_id, label, path, cleanup_owner, timeout=20): size = du_size(path, timeout) if os.path.exists(path) else None return { "id": source_id, "label": label, "path": path, "exists": size is not None, "sizeBytes": size, "sizeHuman": fmt_bytes(size or 0), "cleanupOwner": cleanup_owner, } def pid_alive(pid): try: pid_int = int(pid) except Exception: return False if pid_int <= 0: return False return os.path.exists("/proc/%s" % pid_int) def read_json_file(path): try: with open(path, "r", encoding="utf-8") as handle: value = json.load(handle) return value if isinstance(value, dict) else None except Exception: return None def read_pid_file(path): try: with open(path, "r", encoding="utf-8") as handle: raw = handle.read().strip() return int(raw) if re.match(r"^\d+$", raw) else None except Exception: return None def iso_or_epoch_to_epoch(value): if value is None: return None if isinstance(value, (int, float)): return float(value) text_value = str(value).strip() if not text_value: return None for fmt in ["%Y-%m-%dT%H:%M:%SZ", "%Y-%m-%dT%H:%M:%S.%fZ", "%Y-%m-%d %H:%M:%S"]: try: return float(calendar.timegm(time.strptime(text_value, fmt))) except Exception: pass return None def redact_command_preview(value): text_value = str(value or "") text_value = re.sub(r"(?i)(api[_-]?key|token|authorization|password|secret)=\S+", r"\1=", text_value) text_value = re.sub(r"(?i)(--(?:api-key|token|password|secret))\s+\S+", r"\1 ", text_value) return text_value[:180] def collect_process_pressure(patterns): result = command(["ps", "-eo", "pid=,ppid=,rss=,comm=,args="], 10) if result["exitCode"] != 0: return { "ok": False, "error": "ps-failed", "command": bounded(result), "processCount": 0, "rssBytes": 0, "rows": [], } lowered = [(pattern, pattern.lower()) for pattern in patterns] rows = [] by_pattern = {} for line in result["stdout"].splitlines(): parts = line.strip().split(None, 4) if len(parts) < 4: continue pid, ppid, rss_kib, comm = parts[:4] args = parts[4] if len(parts) >= 5 else comm haystack = ("%s %s" % (comm, args)).lower() matches = [original for original, lowered_pattern in lowered if lowered_pattern and lowered_pattern in haystack] if not matches: continue rss_bytes = safe_int(rss_kib) * 1024 row = { "pid": safe_int(pid), "ppid": safe_int(ppid), "comm": comm, "rssBytes": rss_bytes, "rssHuman": fmt_bytes(rss_bytes), "matchedPatterns": matches, "commandPreview": redact_command_preview(args), } rows.append(row) for pattern in matches: bucket = by_pattern.setdefault(pattern, {"processCount": 0, "rssBytes": 0, "rssHuman": "0 B"}) bucket["processCount"] += 1 bucket["rssBytes"] += rss_bytes bucket["rssHuman"] = fmt_bytes(bucket["rssBytes"]) rows.sort(key=lambda item: safe_int(item.get("rssBytes")), reverse=True) total = sum(safe_int(item.get("rssBytes")) for item in rows) return { "ok": True, "patterns": patterns, "processCount": len(rows), "rssBytes": total, "rssHuman": fmt_bytes(total), "byPattern": by_pattern, "top": rows[:int(OPTIONS.get("limit") or 50)], } def collect_memory_snapshot(): result = command(["free", "-b"], 5) if result["exitCode"] != 0: return {"ok": False, "error": "free-failed", "command": bounded(result)} memory = {} for line in result["stdout"].splitlines(): parts = line.split() if parts and parts[0].rstrip(":") == "Mem" and len(parts) >= 7: memory = { "totalBytes": safe_int(parts[1]), "usedBytes": safe_int(parts[2]), "freeBytes": safe_int(parts[3]), "availableBytes": safe_int(parts[6]), "totalHuman": fmt_bytes(parts[1]), "usedHuman": fmt_bytes(parts[2]), "availableHuman": fmt_bytes(parts[6]), } break return {"ok": bool(memory), "memory": memory, "command": bounded(result)} def observe_run_record(path, stale_hours): stat = os.stat(path) heartbeat = read_json_file(os.path.join(path, "heartbeat.json")) or {} manifest = read_json_file(os.path.join(path, "manifest.json")) or {} pid = None for candidate in ["pid", "observer.pid", "browser.pid", "runner.pid"]: pid = read_pid_file(os.path.join(path, candidate)) if pid is not None: break if pid is None: for source in [heartbeat, manifest]: for key in ["pid", "processId", "runnerPid", "browserPid"]: if source.get(key) is not None: try: pid = int(source.get(key)) break except Exception: pass if pid is not None: break timestamp = None for source in [heartbeat, manifest]: for key in ["updatedAt", "completedAt", "finishedAt", "stoppedAt", "startedAt", "createdAt"]: timestamp = iso_or_epoch_to_epoch(source.get(key)) if timestamp is not None: break if timestamp is not None: break if timestamp is None: timestamp = stat.st_mtime age_hours = max(0.0, (time.time() - timestamp) / 3600.0) status = heartbeat.get("status") or manifest.get("status") or manifest.get("state") alive = pid_alive(pid) terminal = str(status or "").lower() in set(["done", "completed", "complete", "failed", "blocked", "timeout", "timed-out", "stopped", "exited"]) stale_signal = (not alive) and age_hours >= float(stale_hours) and (terminal or status is None) return { "id": os.path.basename(path), "path": path, "pid": pid, "pidAlive": alive, "status": status, "ageHours": round(age_hours, 2), "timestampBasis": "manifest-or-heartbeat" if heartbeat or manifest else "directory-mtime-fallback", "staleSignal": stale_signal, "classification": "review-only", } def collect_web_observe_summary(): roots = config_list(MEMORY_CONFIG, "observeStateRoots", config_list(MEMORY_CONFIG, "webObserveRoots", [])) stale_hours = config_float(MEMORY_CONFIG, "staleRunMaxAgeHours", 6.0, minimum=0.0) if not roots: return { "ok": True, "skipped": True, "reason": "no-yaml-observe-roots", "configSource": "config/unidesk-cli.yaml#gc.remote.targets.%s.memoryPressure.observeStateRoots" % PROVIDER_ID, } root_rows = [] stale_rows = [] active_rows = [] run_count = 0 total_bytes = 0 for root in roots: exists = os.path.isdir(root) root_size = du_size(root, 15) if exists else None if root_size is not None: total_bytes += safe_int(root_size) row = { "root": root, "exists": exists, "sizeBytes": root_size, "sizeHuman": fmt_bytes(root_size or 0), "runCount": 0, "staleSignalCount": 0, "activeSignalCount": 0, } if exists: try: children = [os.path.join(root, name) for name in os.listdir(root)] except OSError: children = [] for child in children: if not os.path.isdir(child): continue try: record = observe_run_record(child, stale_hours) except OSError: continue row["runCount"] += 1 run_count += 1 if record.get("pidAlive"): row["activeSignalCount"] += 1 active_rows.append(record) if record.get("staleSignal"): row["staleSignalCount"] += 1 stale_rows.append(record) root_rows.append(row) stale_rows.sort(key=lambda item: float(item.get("ageHours") or 0), reverse=True) active_rows.sort(key=lambda item: safe_int(item.get("pid"))) return { "ok": True, "configSource": "config/unidesk-cli.yaml#gc.remote.targets.%s.memoryPressure" % PROVIDER_ID, "staleRunMaxAgeHours": stale_hours, "rootCount": len(root_rows), "totalBytes": total_bytes, "totalHuman": fmt_bytes(total_bytes), "runCount": run_count, "activeSignalCount": len(active_rows), "staleSignalCount": len(stale_rows), "roots": root_rows, "activeSignals": active_rows[:int(OPTIONS.get("limit") or 50)], "staleSignals": stale_rows[:int(OPTIONS.get("limit") or 50)], "policy": "analysis-only; active or stale observe runs must be stopped/retained through controlled observer lifecycle commands, not raw process kill or directory deletion", } # __UNIDESK_GC_REMOTE_WEB_OBSERVE_HELPERS__ def collect_memory_pressure(): patterns = config_list(MEMORY_CONFIG, "processPatterns", []) if not patterns: return { "ok": True, "skipped": True, "reason": "no-yaml-process-patterns", "configSource": "config/unidesk-cli.yaml#gc.remote.targets.%s.memoryPressure.processPatterns" % PROVIDER_ID, } processes = collect_process_pressure(patterns) observe = collect_web_observe_summary() return { "ok": processes.get("ok") is True, "configSource": "config/unidesk-cli.yaml#gc.remote.targets.%s.memoryPressure" % PROVIDER_ID, "hostMemory": collect_memory_snapshot(), "processes": processes, "webObserve": observe, "summary": { "matchedProcessCount": processes.get("processCount"), "matchedRssBytes": processes.get("rssBytes"), "matchedRssHuman": processes.get("rssHuman"), "chromeProcessCount": (processes.get("byPattern") or {}).get("chrome", {}).get("processCount"), "observerRunCount": observe.get("runCount"), "activeObserverSignals": observe.get("activeSignalCount"), "staleObserverSignals": observe.get("staleSignalCount"), "observeStateBytes": observe.get("totalBytes"), "observeStateHuman": observe.get("totalHuman"), }, "drillDown": { "processes": "bun scripts/cli.ts gc remote %s snapshot --full --no-save" % PROVIDER_ID, "status": "bun scripts/cli.ts gc remote %s status --job-id " % PROVIDER_ID, }, } def disk_source_snapshot(): sources = [ source_size_item("hwlab-host-data", "HWLAB host data", "/var/lib/hwlab", "hwlab-registry-retention", 60), source_size_item("hwlab-registry", "HWLAB registry", REGISTRY_ROOT, "gc-remote-hwlab-registry", 60), source_size_item("k3s-storage", "k3s local-path storage", "/var/lib/rancher/k3s/storage", "owner-aware-pvc-retention", 45), source_size_item("k3s-containerd", "k3s containerd", "/var/lib/rancher/k3s/agent/containerd", "observation-only", 45), source_size_item("host-containerd", "host containerd", "/var/lib/containerd", "observation-only", 30), source_size_item("kubelet", "kubelet state", "/var/lib/kubelet", "protected-runtime", 20), source_size_item("var-log", "host logs", "/var/log", "gc-remote-logs-journald", 20), source_size_item("tmp", "allowlisted tmp and other tmp", "/tmp", "gc-remote-tmp-allowlist", 20), source_size_item("apt-cache", "apt archives", "/var/cache/apt/archives", "gc-remote-apt-cache", 10), source_size_item("hwlab-v02-source", "HWLAB v0.2 source workspace", "/root/hwlab-v02", "protected-source", 20), source_size_item("agentrun-source", "AgentRun source workspace", "/root/agentrun-v01", "protected-source", 20), ] return [item for item in sources if item.get("exists")] def containerd_breakdown_snapshot(): rows = [ source_size_item("k3s-containerd-content", "k3s containerd content store", "/var/lib/rancher/k3s/agent/containerd/io.containerd.content.v1.content", "observation-only", 30), source_size_item("k3s-containerd-overlayfs", "k3s containerd overlay snapshots", "/var/lib/rancher/k3s/agent/containerd/io.containerd.snapshotter.v1.overlayfs", "observation-only", 30), source_size_item("host-containerd-content", "host containerd content store", "/var/lib/containerd/io.containerd.content.v1.content", "observation-only", 20), source_size_item("host-containerd-overlayfs", "host containerd overlay snapshots", "/var/lib/containerd/io.containerd.snapshotter.v1.overlayfs", "observation-only", 20), ] rows = [item for item in rows if item.get("exists")] return { "state": "observation-only", "cleanupSupported": bool(CONTAINERD_CONFIG), "reason": "k3s image cache cleanup requires --include-k3s-image-cache and uses CRI prune only; host containerd remains observation-only", "breakdown": rows, } # __UNIDESK_GC_REMOTE_CONTAINERD_HELPERS__ # __UNIDESK_GC_REMOTE_PVC_HELPERS__ def compact_memory_pressure(payload): if bool(OPTIONS.get("full")): return payload observe = payload.get("webObserve") or {} return { "ok": payload.get("ok"), "configSource": payload.get("configSource"), "hostMemory": (payload.get("hostMemory") or {}).get("memory"), "webObserve": { "rootCount": observe.get("rootCount"), "totalBytes": observe.get("totalBytes"), "totalHuman": observe.get("totalHuman"), "runCount": observe.get("runCount"), "activeSignalCount": observe.get("activeSignalCount"), "staleSignalCount": observe.get("staleSignalCount"), }, "summary": payload.get("summary"), "drillDown": payload.get("drillDown"), "compacted": True, } def compact_memory_summary(payload): observe = payload.get("webObserve") or {} return { "ok": payload.get("ok"), "configSource": payload.get("configSource"), "summary": payload.get("summary"), "webObserve": { "rootCount": observe.get("rootCount"), "totalBytes": observe.get("totalBytes"), "totalHuman": observe.get("totalHuman"), "runCount": observe.get("runCount"), "activeSignalCount": observe.get("activeSignalCount"), "staleSignalCount": observe.get("staleSignalCount"), }, "compacted": True, "drillDown": "bun scripts/cli.ts gc remote %s status --limit %s" % (PROVIDER_ID, int(OPTIONS.get("limit") or 50)), } # __UNIDESK_GC_REMOTE_GROWTH_HELPERS__ def allocated_file_size(path): try: stat = os.stat(path) blocks = getattr(stat, "st_blocks", 0) if blocks: return int(blocks) * 512 return int(stat.st_size) except OSError: return 0 def df_snapshot(): result = command(["df", "-B1", "-P", "/"], 5) if result["exitCode"] != 0: return None lines = result["stdout"].strip().splitlines() if len(lines) < 2: return None parts = lines[1].split() if len(parts) < 6: return None return { "filesystem": parts[0], "sizeBytes": int(parts[1]), "usedBytes": int(parts[2]), "availableBytes": int(parts[3]), "dfBasisBytes": int(parts[2]) + int(parts[3]), "reservedBytes": max(0, int(parts[1]) - int(parts[2]) - int(parts[3])), "usePercentExact": round((int(parts[2]) * 100.0 / (int(parts[2]) + int(parts[3]))) if (int(parts[2]) + int(parts[3])) > 0 else 0.0, 2), "usePercent": int(parts[4].replace("%", "")), "mount": parts[5], } def fmt_bytes(value): units = ["B", "KiB", "MiB", "GiB", "TiB"] size = float(max(0, int(value or 0))) idx = 0 while size >= 1024 and idx < len(units) - 1: size /= 1024.0 idx += 1 return ("%0.0f %s" if size >= 10 or idx == 0 else "%0.1f %s") % (size, units[idx]) def disk_use_percent(size_bytes, used_bytes): try: size = int(size_bytes or 0) used = int(used_bytes or 0) except Exception: return None if size <= 0: return None return int((max(0, used) * 100 + size - 1) // size) def parse_journal_usage(text): m = re.search(r"take up\s+([0-9.]+)\s*([KMGT]?)(?:i?B|B)?", text, re.I) if not m: return None mult = {"": 1, "K": 1024, "M": 1024**2, "G": 1024**3, "T": 1024**4}.get(m.group(2).upper(), 1) return int(float(m.group(1)) * mult) def parse_docker_human_size(raw): raw = str(raw).split("(")[0].strip() m = re.match(r"^([0-9.]+)\s*([KMGT]?B)$", raw, re.I) if not m: return None mult = {"B": 1, "KB": 1000, "MB": 1000**2, "GB": 1000**3, "TB": 1000**4}.get(m.group(2).upper(), 1) return int(float(m.group(1)) * mult) def parse_docker_build_cache(text): for line in text.splitlines(): if not line.startswith("Build Cache"): continue match = re.match(r"^Build Cache\s+\S+\s+\S+\s+(\S+)\s+(\S+)", line.strip()) if not match: continue size = parse_docker_human_size(match.group(1)) reclaim = parse_docker_human_size(match.group(2)) if size is None or reclaim is None: return None return {"sizeBytes": size, "reclaimableBytes": reclaim} return None def docker_containers(): ps = command(["docker", "ps", "-qa", "--no-trunc"], 5) if ps["exitCode"] != 0 or not ps["stdout"].strip(): return [] ids = ps["stdout"].split() inspect = command(["docker", "inspect"] + ids, 10) if inspect["exitCode"] != 0 or not inspect["stdout"].strip(): return [] try: data = json.loads(inspect["stdout"]) except Exception: return [] rows = [] for item in data: cfg = item.get("Config") or {} rows.append({ "id": str(item.get("Id") or ""), "name": str(item.get("Name") or "").lstrip("/"), "image": str(cfg.get("Image") or item.get("Image") or ""), "logPath": str(item.get("LogPath") or ""), }) return [row for row in rows if row["id"]] def cluster_preflight(): node_cmd = command(["sh", "-lc", "KUBECONFIG=/etc/rancher/k3s/k3s.yaml kubectl get nodes -o jsonpath='{range .items[*]}{.metadata.name}{\"\\n\"}{end}' 2>/dev/null"], 10) pods_cmd = command(["sh", "-lc", "KUBECONFIG=/etc/rancher/k3s/k3s.yaml kubectl get pods -n hwlab-dev --no-headers 2>/dev/null | wc -l"], 10) nodes = [line.strip() for line in node_cmd["stdout"].splitlines() if line.strip()] expected = EXPECTED_G14_NODE if PROVIDER_ID.upper() == "G14" else None ok = True reason = "ok" if expected is not None and expected not in nodes: ok = False reason = "expected-g14-node-missing" return { "ok": ok, "reason": reason, "providerId": PROVIDER_ID, "hostname": command(["hostname"], 5)["stdout"].strip(), "expectedNode": expected, "nodes": nodes, "nodeCommand": bounded(node_cmd), "hwlabDevPodCount": int(pods_cmd["stdout"].strip() or "0") if pods_cmd["exitCode"] == 0 else None, "hwlabDevPodCommand": bounded(pods_cmd), } # __UNIDESK_GC_REMOTE_REGISTRY_HELPERS__ def collect_protected(): protected_paths = [ ("hwlab-k3s-runtime", "/var/lib/rancher/k3s", "Native k3s runtime, containerd state, local-path storage and control-plane data are protected."), ("hwlab-k3s-storage", "/var/lib/rancher/k3s/storage", "Local-path PVC data is protected; cleanup must go through HWLAB/Tekton retention commands."), ("hwlab-kubelet", "/var/lib/kubelet", "Kubelet pod/runtime state is protected."), ("host-containerd", "/var/lib/containerd", "Host containerd state is protected; generic gc does not prune containerd images."), ("hwlab-host-data", "/var/lib/hwlab", "HWLAB host data/cache is protected from generic remote gc until a HWLAB-specific retention rule classifies it."), ("hwlab-source", "/root/hwlab", "G14 HWLAB fixed source workspace is protected."), ("hwlab-v02-source", "/root/hwlab-v02", "HWLAB v0.2 fixed source workspace is protected."), ("agentrun-source", "/root/agentrun", "AgentRun fixed source workspace is protected."), ("docker-images-and-volumes", "docker-images-volumes", "Remote gc does not remove Docker images, containers, volumes or Compose projects."), ("k8s-api-objects", "deployments-statefulsets-secrets-pvcs", "Remote gc does not mutate Kubernetes workloads, Secrets, PVCs, PVs, Argo CD or Tekton objects."), ] if not host_docker_gc_enabled("dockerJsonLogs"): protected_paths.append(("host-docker-json-logs", "/var/lib/docker/containers", host_docker_gc_reason("dockerJsonLogs", "Host Docker json logs are protected on this provider."))) if not host_docker_gc_enabled("buildCachePrune"): protected_paths.append(("host-docker-build-cache", "docker-builder-cache", host_docker_gc_reason("buildCachePrune", "Host Docker builder cache prune is disabled on this provider."))) result = [] for kind, ref, reason in protected_paths: item = {"kind": kind, "risk": "blocked", "ref": ref, "reason": reason} if ref.startswith("/") and os.path.exists(ref): size = du_size(ref, 3, False) item["sizeBytes"] = size item["sizeState"] = "sampled" if size is not None else "timeout-or-unavailable" result.append(item) return result def collect_candidates(observed_at): candidates = [] if OPTIONS.get("journal", True): usage = command(["journalctl", "--disk-usage"], 5) current = parse_journal_usage((usage["stdout"] or "") + (usage["stderr"] or "")) target = int(OPTIONS.get("journalTargetBytes") or 536870912) if current is not None and current > target: candidates.append({ "id": "journalctl:vacuum", "kind": "journal-vacuum", "risk": "medium", "description": "Vacuum systemd journal to %s" % fmt_bytes(target), "sizeBytes": current, "estimatedReclaimBytes": max(0, current - target), "action": {"command": ["journalctl", "--vacuum-size=%s" % target]}, }) if OPTIONS.get("dockerLogs", True) and host_docker_gc_enabled("dockerJsonLogs"): max_bytes = int(OPTIONS.get("dockerLogMaxBytes") or 52428800) for container in docker_containers(): path = container.get("logPath") or "" if not path or not os.path.exists(path): continue try: size = os.path.getsize(path) except OSError: continue if size <= max_bytes: continue candidates.append({ "id": "docker-json-log:%s" % container["id"], "kind": "docker-json-log-truncate", "risk": "medium", "description": "Truncate Docker json-file log larger than %s" % fmt_bytes(max_bytes), "path": path, "container": {"id": container["id"][:12], "name": container["name"], "image": container["image"]}, "sizeBytes": size, "estimatedReclaimBytes": size, "action": {"op": "truncate", "targetBytes": 0}, }) if OPTIONS.get("buildCache", True) and host_docker_gc_enabled("buildCachePrune"): system_df = command(["docker", "system", "df"], 8) if system_df["exitCode"] == 0: cache = parse_docker_build_cache(system_df["stdout"]) if cache is not None and cache["reclaimableBytes"] > 0: until = str(OPTIONS.get("buildCacheUntil") or "24h") candidates.append({ "id": "docker-builder:prune", "kind": "docker-build-cache-prune", "risk": "low", "description": "Prune Docker BuildKit cache unused for %s" % until, "sizeBytes": cache["sizeBytes"], "estimatedReclaimBytes": cache["reclaimableBytes"], "action": {"command": ["docker", "builder", "prune", "--all", "--force", "--filter", "until=%s" % until], "estimate": "docker-system-df-reclaimable-upper-bound"}, }) if OPTIONS.get("aptCache", True): apt_path = "/var/cache/apt/archives" size = du_size(apt_path) or 0 if size > 10 * 1024 * 1024: candidates.append({ "id": "apt-cache:clean", "kind": "apt-cache-clean", "risk": "low", "description": "Clean downloaded apt package archives", "path": apt_path, "sizeBytes": size, "estimatedReclaimBytes": size, "action": {"command": ["apt-get", "clean"]}, }) if OPTIONS.get("toolCaches", False): for item in TOOL_CACHE_ALLOWLIST: path = item["path"] size = du_size(path, 8) or 0 if size <= 0: continue candidates.append({ "id": "tool-cache:%s" % item["id"], "kind": "tool-cache-delete", "risk": "medium", "description": item["description"], "path": path, "sizeBytes": size, "estimatedReclaimBytes": size, "action": {"op": "rm-recursive", "allowlist": "remote-tool-cache"}, }) if OPTIONS.get("webObserveArtifacts", False): observe = collect_web_observe_summary() for record in observe.get("staleSignals") or []: path = record.get("path") or "" if not path or not is_direct_observe_run_path(path): continue try: if path_has_open_fd(path): continue size = du_size(path, 15) or path_size(path) except Exception: continue if size <= 0: continue candidates.append({ "id": "web-observe-run:%s" % record.get("id"), "kind": "web-observe-run-delete", "risk": "medium", "description": "Delete stale web-observe artifact run under YAML-configured observeStateRoots", "path": path, "sizeBytes": size, "estimatedReclaimBytes": size, "stale": { "id": record.get("id"), "ageHours": record.get("ageHours"), "timestampBasis": record.get("timestampBasis"), "pid": record.get("pid"), "pidAlive": record.get("pidAlive"), "status": record.get("status"), }, "action": {"op": "rm-recursive", "allowlist": "web-observe-stale-run"}, }) if OPTIONS.get("k3sImageCache", False): candidate = k3s_image_cache_candidate() if candidate: candidates.append(candidate) if OPTIONS.get("hostContainerdCache", False): candidate = host_containerd_cache_candidate() if candidate: candidates.append(candidate) if OPTIONS.get("localPathOrphans", False): candidate = local_path_orphan_candidate() if candidate: candidates.append(candidate) if OPTIONS.get("coreDumps", True): cutoff = time.time() - float(OPTIONS.get("coreDumpMinAgeHours") or 1) * 3600 for root in sorted(CORE_DUMP_DIR_ALLOWLIST): if not os.path.isdir(root): continue for name in os.listdir(root): if not re.match(r"^core\.\d+$", name): continue path = os.path.join(root, name) try: stat = os.lstat(path) except OSError: continue if not os.path.isfile(path) or os.path.islink(path): continue if stat.st_mtime >= cutoff: continue disk_size = allocated_file_size(path) candidates.append({ "id": "core-dump:%s" % path, "kind": "core-dump-delete", "risk": "low", "description": "Delete untracked process core dump older than %s hours" % OPTIONS.get("coreDumpMinAgeHours"), "path": path, "sizeBytes": int(disk_size), "estimatedReclaimBytes": int(disk_size), "apparentSizeBytes": int(stat.st_size), "action": {"op": "unlink", "allowlist": "root-unidesk-core-dot-pid"}, }) if OPTIONS.get("tmp", True) and os.path.isdir("/tmp"): cutoff = time.time() - float(OPTIONS.get("tmpMinAgeHours") or 24) * 3600 for name in os.listdir("/tmp"): path = os.path.join("/tmp", name) if path in TMP_EXACT_PROTECT: continue if not any(name.startswith(prefix) for prefix in TMP_PREFIX_ALLOWLIST): continue try: stat = os.lstat(path) except OSError: continue if stat.st_mtime >= cutoff: continue size = du_size(path, 8) or path_size(path) if size <= 0: continue candidates.append({ "id": "tmp:%s" % path, "kind": "tmp-path-delete", "risk": "low", "description": "Delete allowlisted /tmp path older than %s hours" % OPTIONS.get("tmpMinAgeHours"), "path": path, "sizeBytes": size, "estimatedReclaimBytes": size, "action": {"op": "rm-recursive", "allowlist": "tmp-prefix"}, }) if OPTIONS.get("hwlabRegistry", False): registry = plan_registry_retention() summary = registry.get("summary") or {} delete_rows = registry.get("deleteRows") or [] delete_revision_rows = registry.get("deleteRevisionRows") or [] estimate = int(summary.get("estimatedReclaimBytes") or 0) if delete_rows or delete_revision_rows: candidates.append({ "id": "hwlab-registry:retention-gc", "kind": "hwlab-registry-retention-gc", "risk": "medium", "description": "Conservative HWLAB registry retention: keep current workload refs, retained tags and protected repos, delete stale manifest revisions, then run official registry garbage-collect", "path": REGISTRY_ROOT, "sizeBytes": int(summary.get("registrySizeBytes") or 0), "estimatedReclaimBytes": estimate, "action": { "op": "registry-retention-gc", "requiresMaintenanceWindow": True, "keepPerRepo": summary.get("keepPerRepo"), "minAgeHours": summary.get("minAgeHours"), "deleteTags": len(delete_rows), "deleteManifests": summary.get("deleteManifests"), "deleteRevisions": summary.get("deleteRevisions"), "deleteByRepo": summary.get("deleteByRepo"), "revisionDeleteByRepo": summary.get("revisionDeleteByRepo"), "protectedWorkloadRefs": summary.get("protectedWorkloadRefs"), "protectedDigestRefs": summary.get("protectedDigestRefs"), "protectedDigestClosure": summary.get("protectedDigestClosure"), }, }) elif bool(OPTIONS.get("registryGcOnly")) and int(summary.get("totalTags") or 0) > 0 and int(summary.get("deleteTags") or 0) == 0: candidates.append({ "id": "hwlab-registry:garbage-collect-only", "kind": "hwlab-registry-garbage-collect", "risk": "medium", "description": "Run official HWLAB registry garbage-collect without deleting additional tags; useful after a previously interrupted retention run", "path": REGISTRY_ROOT, "sizeBytes": int(summary.get("registrySizeBytes") or 0), "estimatedReclaimBytes": 0, "action": { "op": "registry-garbage-collect-only", "requiresMaintenanceWindow": True, "deleteTags": 0, "registryPlan": summary, }, }) return sorted(candidates, key=lambda item: item.get("estimatedReclaimBytes") or 0, reverse=True) def target_assessment(disk, estimated_reclaim): raw = OPTIONS.get("targetUsePercent") if raw is None: return None if not disk: return { "targetUsePercent": raw, "ok": False, "state": "unavailable", "reason": "disk-snapshot-unavailable", } try: target = int(raw) size = int(disk.get("sizeBytes") or 0) used = int(disk.get("usedBytes") or 0) available = int(disk.get("availableBytes") or 0) reclaim = max(0, int(estimated_reclaim or 0)) except Exception: return { "targetUsePercent": raw, "ok": False, "state": "unavailable", "reason": "invalid-disk-snapshot", } df_basis = used + available if df_basis <= 0: df_basis = size legacy_target_used_bytes = (size * target) // 100 legacy_required = max(0, used - legacy_target_used_bytes) target_used_bytes = (df_basis * target) // 100 required = max(0, used - target_used_bytes) projected_used = max(0, used - reclaim) projected_use_percent = disk_use_percent(df_basis, projected_used) legacy_projected_use_percent = disk_use_percent(size, projected_used) enough = reclaim >= required if required == 0: state = "already-below-target" elif enough: state = "candidate-estimate-meets-target" elif reclaim <= 0: state = "safe-stop-no-meaningful-candidates" else: state = "shortfall" shortfall = max(0, required - reclaim) return { "targetUsePercent": target, "ok": required == 0 or enough, "state": state, "currentUsePercent": disk.get("usePercent"), "currentUsePercentExact": disk.get("usePercentExact"), "basis": "df-used-over-used-plus-available", "dfBasisBytes": df_basis, "dfBasis": fmt_bytes(df_basis), "reservedBytes": max(0, size - df_basis), "reserved": fmt_bytes(max(0, size - df_basis)), "currentUsedBytes": used, "currentUsed": fmt_bytes(used), "targetUsedBytes": target_used_bytes, "targetUsed": fmt_bytes(target_used_bytes), "requiredReclaimBytes": required, "requiredReclaim": fmt_bytes(required), "estimatedReclaimBytes": reclaim, "estimatedReclaim": fmt_bytes(reclaim), "shortfallBytes": shortfall, "shortfall": fmt_bytes(shortfall), "projectedUsedBytes": projected_used, "projectedUsed": fmt_bytes(projected_used), "projectedUsePercent": projected_use_percent, "safeStop": required > 0 and not enough, "decision": "stop-and-escalate-retention-or-capacity" if required > 0 and not enough else "target-covered-by-safe-candidates", "legacySizeBasis": { "basis": "df-size-column-includes-reserved-blocks", "sizeBytes": size, "size": fmt_bytes(size), "targetUsedBytes": legacy_target_used_bytes, "targetUsed": fmt_bytes(legacy_target_used_bytes), "requiredReclaimBytes": legacy_required, "requiredReclaim": fmt_bytes(legacy_required), "projectedUsePercent": legacy_projected_use_percent, "note": "informational only; ok/safeStop use the same basis as df Use%", }, } def summarize(candidates, returned, disk=None): by_kind = {} total = 0 for item in candidates: size = int(item.get("estimatedReclaimBytes") or 0) total += size kind = item.get("kind") or "unknown" current = by_kind.setdefault(kind, {"count": 0, "estimatedReclaimBytes": 0, "estimatedReclaim": "0 B"}) current["count"] += 1 current["estimatedReclaimBytes"] += size current["estimatedReclaim"] = fmt_bytes(current["estimatedReclaimBytes"]) returned_total = sum(int(item.get("estimatedReclaimBytes") or 0) for item in returned) return { "candidateCount": len(candidates), "returnedCandidateCount": len(returned), "estimatedReclaimBytes": total, "estimatedReclaim": fmt_bytes(total), "returnedEstimatedReclaimBytes": returned_total, "returnedEstimatedReclaim": fmt_bytes(returned_total), "byKind": by_kind, "target": target_assessment(disk, total), } def assert_tmp_candidate(path): resolved = os.path.abspath(path) if not resolved.startswith("/tmp/"): raise RuntimeError("refusing to remove non-/tmp path: %s" % path) if resolved in TMP_EXACT_PROTECT: raise RuntimeError("refusing to remove protected tmp path: %s" % path) name = os.path.basename(resolved) if not any(name.startswith(prefix) for prefix in TMP_PREFIX_ALLOWLIST): raise RuntimeError("refusing to remove tmp path outside allowlist: %s" % path) def assert_core_dump_candidate(path): resolved = os.path.abspath(path) parent = os.path.dirname(resolved) name = os.path.basename(resolved) if parent not in CORE_DUMP_DIR_ALLOWLIST: raise RuntimeError("refusing to remove core dump outside allowlisted directory: %s" % path) if not re.match(r"^core\.\d+$", name): raise RuntimeError("refusing to remove non core. file: %s" % path) if not os.path.isfile(resolved) or os.path.islink(resolved): raise RuntimeError("refusing to remove non-regular core dump: %s" % path) git = command(["git", "-C", parent, "ls-files", "--error-unmatch", name], 10) if git["exitCode"] == 0: raise RuntimeError("refusing to remove git-tracked file: %s" % path) if git["exitCode"] != 1: raise RuntimeError("refusing to remove core dump because git tracking check failed: %s" % path) fuser = command(["fuser", resolved], 5) if fuser["exitCode"] is None: raise RuntimeError("refusing to remove core dump because fuser check was unavailable: %s" % path) if fuser["exitCode"] == 0: raise RuntimeError("refusing to remove core dump with active process reference: %s" % path) def assert_tool_cache_candidate(path): resolved = os.path.abspath(path) allowed = set(item["path"] for item in TOOL_CACHE_ALLOWLIST) if resolved not in allowed: raise RuntimeError("refusing to remove tool cache outside allowlist: %s" % path) if os.path.islink(resolved): raise RuntimeError("refusing to remove symlink tool cache: %s" % path) def execute(candidate): kind = candidate.get("kind") if kind == "journal-vacuum": result = command(["journalctl", "--vacuum-size=%s" % int(OPTIONS.get("journalTargetBytes") or 536870912)], 30) if result["exitCode"] != 0: raise RuntimeError((result["stderr"] or "journalctl vacuum failed").strip()) return {"reclaimedBytes": None, "commandOutput": bounded(result)} if kind == "docker-json-log-truncate": if not host_docker_gc_enabled("dockerJsonLogs"): raise RuntimeError(host_docker_gc_reason("dockerJsonLogs", "refusing Docker json log truncation on this provider")) path = candidate.get("path") or "" if not path.startswith("/var/lib/docker/containers/"): raise RuntimeError("refusing to truncate Docker log outside /var/lib/docker/containers") before = os.path.getsize(path) if os.path.exists(path) else 0 with open(path, "r+b") as handle: handle.truncate(0) return {"reclaimedBytes": before} if kind == "docker-build-cache-prune": if not host_docker_gc_enabled("buildCachePrune"): raise RuntimeError(host_docker_gc_reason("buildCachePrune", "refusing Docker builder prune on this provider")) until = str(OPTIONS.get("buildCacheUntil") or "24h") result = command(["docker", "builder", "prune", "--all", "--force", "--filter", "until=%s" % until], 45) if result["exitCode"] != 0: raise RuntimeError((result["stderr"] or "docker builder prune failed").strip()) return {"reclaimedBytes": None, "commandOutput": bounded(result)} if kind == "apt-cache-clean": before = du_size("/var/cache/apt/archives") or 0 result = command(["apt-get", "clean"], 30) if result["exitCode"] != 0: raise RuntimeError((result["stderr"] or "apt-get clean failed").strip()) after = du_size("/var/cache/apt/archives") or 0 return {"reclaimedBytes": max(0, before - after), "commandOutput": bounded(result)} if kind == "tool-cache-delete": path = candidate.get("path") or "" assert_tool_cache_candidate(path) before = du_size(path, 8) or path_size(path) if os.path.isdir(path): shutil.rmtree(path, ignore_errors=True) elif os.path.exists(path): os.unlink(path) return {"reclaimedBytes": before} if kind == "web-observe-run-delete": path = candidate.get("path") or "" record = assert_web_observe_candidate(path) before = du_size(path, 15) or path_size(path) shutil.rmtree(path, ignore_errors=True) return {"reclaimedBytes": before, "stale": record} if kind == "k3s-cri-image-prune": return execute_k3s_image_cache_prune() if kind == "host-containerd-cache-prune": return execute_host_containerd_cache_prune() if kind == "host-containerd-orphan-state-delete": return execute_host_containerd_orphan_cleanup() if kind == "k3s-local-path-orphans-delete": return execute_local_path_orphan_cleanup() if kind == "tmp-path-delete": path = candidate.get("path") or "" assert_tmp_candidate(path) before = du_size(path, 8) or path_size(path) if os.path.isdir(path) and not os.path.islink(path): shutil.rmtree(path, ignore_errors=True) else: try: os.unlink(path) except FileNotFoundError: pass return {"reclaimedBytes": before} if kind == "core-dump-delete": path = candidate.get("path") or "" assert_core_dump_candidate(path) before = allocated_file_size(path) os.unlink(path) return {"reclaimedBytes": before} if kind == "hwlab-registry-retention-gc": return start_registry_retention_job("retention") if kind == "hwlab-registry-garbage-collect": return start_registry_retention_job("garbage-collect") raise RuntimeError("unsupported remote gc candidate kind: %s" % kind) def visible_items(items): if bool(OPTIONS.get("full")): return items return items[:int(OPTIONS.get("limit") or 50)] def returned_results(results): def compact_result(item): if bool(OPTIONS.get("full")): return item return { key: item.get(key) for key in [ "id", "kind", "risk", "path", "status", "estimatedReclaimBytes", "reclaimedBytes", "orphanCount", "selectedOrphanCount", "overlayCandidateCount", "contentCandidateCount", "deletedOrphanCount", "error", ] if item.get(key) is not None } if bool(OPTIONS.get("full")): return results failed = [item for item in results if item.get("status") == "failed"] started = [item for item in results if item.get("status") == "started"] succeeded = [item for item in results if item.get("status") == "succeeded"] return [compact_result(item) for item in (failed + started + succeeded)[:int(OPTIONS.get("resultLimit") or 50)]] def plan_payload(observed_at, preflight, protected, candidates, visible): disk = df_snapshot() ci_storage = ci_storage_snapshot() memory_pressure = collect_memory_pressure() compact_pvc = compact_pvc_attribution(ci_storage) policy = { "requiresRunConfirm": True, "runCommand": "bun scripts/cli.ts gc remote %s run --confirm" % PROVIDER_ID, "neverTouches": [ "/var/lib/rancher/k3s", "/var/lib/rancher/k3s/storage", "/var/lib/kubelet", "/var/lib/containerd", "/var/lib/hwlab unless --include-hwlab-registry is explicitly supplied", "Kubernetes Deployments/StatefulSets/Secrets/PVCs/PVs", "HWLAB fixed source workspaces", "Docker images, containers and volumes", ], "notes": [ "Remote gc only executes the returned candidate page unless --full or a larger --limit is supplied.", "G14 run requires the expected native k3s node preflight before mutation.", "HWLAB DEV runtime and local-path PVC data are protected and require HWLAB-specific retention commands.", "Core dump cleanup only removes untracked /root/unidesk/core. regular files with no active fuser reference.", "HWLAB registry retention is opt-in: it keeps workload tag/digest refs, all tags newer than the retention age and the newest N tags per repo before official registry garbage-collect.", "When summary.target.safeStop is true, do not broaden deletion scope; choose registry retention, k3s/containerd image cache maintenance, PVC/runtime retention or capacity expansion explicitly.", ], } if not bool(OPTIONS.get("full")): policy = { "requiresRunConfirm": True, "runCommand": "bun scripts/cli.ts gc remote %s run --confirm" % PROVIDER_ID, "neverTouches": ["k3s runtime", "PVC/PV/local-path data", "Secrets/auth/config", "Docker volumes/images"], "notes": [ "Default plan is compact; rerun with --full for complete policy notes and protected rows.", "When summary.target.safeStop is true, stop at protected boundaries and choose an owner-aware retention or capacity decision.", ], } payload = { "ok": True, "action": "gc remote plan", "providerId": PROVIDER_ID, "dryRun": True, "mutation": False, "observedAt": observed_at, "options": OPTIONS, "diskBefore": disk, "clusterPreflight": preflight, "summary": summarize(candidates, visible, disk), "candidates": visible, "protected": protected if bool(OPTIONS.get("full")) else protected[:3], "policy": policy, } if bool(OPTIONS.get("full")): payload.update({ "memoryPressure": memory_pressure, "pvcAttribution": ci_storage, "ciStorage": ci_storage, }) else: payload["pressureSummary"] = { "memory": (compact_memory_pressure(memory_pressure).get("summary") if isinstance(compact_memory_pressure(memory_pressure), dict) else None), "pvc": { "pvcCount": compact_pvc.get("pvcCount"), "reviewCandidateCount": compact_pvc.get("reviewCandidateCount"), "estimatedBytes": compact_pvc.get("estimatedBytes"), "estimatedHuman": compact_pvc.get("estimatedHuman"), "byNamespace": compact_pvc.get("byNamespace"), "handoff": compact_pvc.get("handoff"), }, "drillDown": "bun scripts/cli.ts gc remote %s status --limit %s" % (PROVIDER_ID, int(OPTIONS.get("limit") or 50)), } return payload def safe_unit_name(value): raw = str(value or "").strip().lower() raw = re.sub(r"[^a-z0-9_.@-]+", "-", raw).strip("-") if not raw: raw = "unidesk-%s-low-risk-gc" % re.sub(r"[^a-z0-9]+", "-", PROVIDER_ID.lower()).strip("-") return raw[:80] def render_remote_policy(): unit_name = safe_unit_name(config_str(POLICY_TIMER_CONFIG, "name", "unidesk-%s-low-risk-gc" % PROVIDER_ID.lower())) on_calendar = config_str(POLICY_TIMER_CONFIG, "onCalendar", "daily") randomized_delay_sec = config_str(POLICY_TIMER_CONFIG, "randomizedDelaySec", "15min") journal_target = parse_size_value(POLICY_TIMER_CONFIG.get("journalTargetBytes"), int(OPTIONS.get("journalTargetBytes") or 536870912)) tmp_min_age_hours = config_float(POLICY_TIMER_CONFIG, "tmpMinAgeHours", float(OPTIONS.get("tmpMinAgeHours") or 24), minimum=0.0) include_apt_cache = config_bool(POLICY_TIMER_CONFIG, "includeAptCache", bool(OPTIONS.get("aptCache", True))) include_tool_caches = config_bool(POLICY_TIMER_CONFIG, "includeToolCaches", False) include_web_observe = config_bool(POLICY_TIMER_CONFIG, "includeWebObserveArtifacts", False) include_k3s_images = config_bool(POLICY_TIMER_CONFIG, "includeK3sImageCache", False) include_host_containerd = config_bool(POLICY_TIMER_CONFIG, "includeHostContainerdCache", False) include_local_path_orphans = config_bool(POLICY_TIMER_CONFIG, "includeLocalPathOrphans", False) state_dir = config_str(POLICY_TIMER_CONFIG, "stateDir", "/var/lib/unidesk-gc") config_dir = config_str(POLICY_TIMER_CONFIG, "configDir", "/etc/unidesk-gc") script_path = "/usr/local/sbin/%s.py" % unit_name config_path = os.path.join(config_dir, "%s.json" % unit_name) state_path = os.path.join(state_dir, "%s.last.json" % unit_name) service_path = "/etc/systemd/system/%s.service" % unit_name timer_path = "/etc/systemd/system/%s.timer" % unit_name policy_config = { "providerId": PROVIDER_ID, "unitName": unit_name, "statePath": state_path, "journalTargetBytes": int(journal_target), "tmpMinAgeHours": tmp_min_age_hours, "tmpPrefixAllowlist": TMP_PREFIX_ALLOWLIST, "tmpExactProtect": sorted(TMP_EXACT_PROTECT), "includeAptCache": include_apt_cache, "toolCachePaths": [item["path"] for item in TOOL_CACHE_ALLOWLIST] if include_tool_caches else [], "webObserve": {"enabled": include_web_observe, **MEMORY_CONFIG}, "k3sImageCache": {"enabled": include_k3s_images, **CONTAINERD_CONFIG}, "hostContainerdCache": {"enabled": include_host_containerd, **HOST_CONTAINERD_CONFIG}, "localPathStorage": {"enabled": include_local_path_orphans, **LOCAL_PATH_CONFIG}, "agentrunSessionPvcs": POLICY_TIMER_CONFIG.get("agentrunSessionPvcs") if isinstance(POLICY_TIMER_CONFIG.get("agentrunSessionPvcs"), dict) else {"enabled": False}, } stages = [ {"id": "journal", "enabled": True, "risk": "low", "mode": "journalctl-vacuum"}, {"id": "apt-cache", "enabled": include_apt_cache, "risk": "low", "mode": "apt-get-clean"}, {"id": "tmp-allowlist", "enabled": True, "risk": "low", "mode": "direct-child-prefix-retention"}, {"id": "tool-caches", "enabled": include_tool_caches, "risk": "medium", "mode": "fixed-path-rebuildable-cache"}, {"id": "web-observe-artifacts", "enabled": include_web_observe, "risk": "medium", "mode": "manifest-heartbeat-pid-openfd-stale-run"}, {"id": "agentrun-session-pvcs", "enabled": bool(policy_config["agentrunSessionPvcs"].get("enabled")), "risk": "medium", "mode": "kubectl-delete-pvc-wait-false-owner-aware"}, {"id": "k3s-image-cache", "enabled": include_k3s_images, "risk": "medium", "mode": "crictl-rmi-prune-ci-idle"}, {"id": "host-containerd-orphans", "enabled": include_host_containerd, "risk": "medium", "mode": "ctr-metadata-empty-yaml-orphan-state"}, {"id": "local-path-orphans", "enabled": include_local_path_orphans, "risk": "medium", "mode": "pv-unreferenced-direct-child"}, ] service = "\n".join([ "[Unit]", "Description=UniDesk remote growth-slowdown GC for %s" % PROVIDER_ID, "Documentation=config/unidesk-cli.yaml#gc.remote.targets.%s.policyTimer" % PROVIDER_ID, "", "[Service]", "Type=oneshot", "ExecStart=/usr/bin/python3 %s %s" % (script_path, config_path), "Nice=10", "IOSchedulingClass=best-effort", "IOSchedulingPriority=7", "", ]) timer = "\n".join([ "[Unit]", "Description=UniDesk remote low-risk GC timer for %s" % PROVIDER_ID, "", "[Timer]", "OnCalendar=%s" % on_calendar, "RandomizedDelaySec=%s" % randomized_delay_sec, "Persistent=true", "", "[Install]", "WantedBy=timers.target", "", ]) return { "unitName": unit_name, "scriptPath": script_path, "configPath": config_path, "statePath": state_path, "configDir": config_dir, "stateDir": state_dir, "servicePath": service_path, "timerPath": timer_path, "onCalendar": on_calendar, "randomizedDelaySec": randomized_delay_sec, "journalTargetBytes": int(journal_target), "journalTarget": fmt_bytes(journal_target), "tmpMinAgeHours": tmp_min_age_hours, "includeAptCache": include_apt_cache, "includeToolCaches": include_tool_caches, "includeWebObserveArtifacts": include_web_observe, "includeK3sImageCache": include_k3s_images, "includeHostContainerdCache": include_host_containerd, "includeLocalPathOrphans": include_local_path_orphans, "stages": stages, "policyConfig": policy_config, "script": POLICY_RUNNER_SOURCE, "service": service, "timer": timer, } def remote_policy_plan_payload(observed_at): rendered = render_remote_policy() return { "ok": True, "action": "gc remote policy plan", "providerId": PROVIDER_ID, "dryRun": True, "mutation": False, "observedAt": observed_at, "configSource": "config/unidesk-cli.yaml#gc.remote.targets.%s.policyTimer" % PROVIDER_ID, "enabled": config_bool(POLICY_TIMER_CONFIG, "enabled", False), "timer": {key: rendered.get(key) for key in ["unitName", "scriptPath", "configPath", "statePath", "servicePath", "timerPath", "onCalendar", "randomizedDelaySec", "journalTargetBytes", "journalTarget", "tmpMinAgeHours", "includeAptCache", "includeToolCaches", "includeWebObserveArtifacts", "includeK3sImageCache", "includeHostContainerdCache", "includeLocalPathOrphans"]}, "stages": rendered.get("stages"), "servicePreview": rendered["service"] if bool(OPTIONS.get("full")) else None, "timerPreview": rendered["timer"] if bool(OPTIONS.get("full")) else None, "installCommand": "bun scripts/cli.ts gc remote %s policy install --confirm" % PROVIDER_ID, "statusCommand": "bun scripts/cli.ts gc remote %s policy status" % PROVIDER_ID, "policy": { "risk": "low-to-medium-owner-aware", "neverTouches": [ "k3s runtime metadata unless a dedicated CRI prune stage is enabled", "active PVC/PV/local-path data", "Docker images, containers, volumes or Docker build cache", "Secret/auth/config state", "active Web observe runners or Chrome processes", ], "sourceOfTruth": "config/unidesk-cli.yaml#gc.remote.targets.%s.policyTimer" % PROVIDER_ID, }, } def remote_policy_status_payload(observed_at): rendered = render_remote_policy() timer = command(["systemctl", "show", "%s.timer" % rendered["unitName"], "--property=LoadState,ActiveState,SubState,NextElapseUSecRealtime,LastTriggerUSec"], 10) service = command(["systemctl", "show", "%s.service" % rendered["unitName"], "--property=LoadState,ActiveState,SubState,Result,ExecMainStatus"], 10) last = None try: with open(rendered["statePath"], "r", encoding="utf-8") as handle: last = json.load(handle) except Exception: last = None if last is not None and not bool(OPTIONS.get("full")): last = compact_policy_last_run(last) return { "ok": True, "action": "gc remote policy status", "providerId": PROVIDER_ID, "dryRun": True, "mutation": False, "observedAt": observed_at, "timer": {key: rendered.get(key) for key in ["unitName", "scriptPath", "configPath", "statePath", "servicePath", "timerPath", "onCalendar", "randomizedDelaySec"]}, "systemd": { "timer": bounded(timer), "service": bounded(service), }, "lastRun": last, "next": { "plan": "bun scripts/cli.ts gc remote %s policy plan" % PROVIDER_ID, "install": "bun scripts/cli.ts gc remote %s policy install --confirm" % PROVIDER_ID, "disk": "bun scripts/cli.ts gc remote %s status --limit 20" % PROVIDER_ID, }, } def compact_policy_last_run(last): results = [] for item in last.get("results") or []: if not isinstance(item, dict): continue compact = { key: item.get(key) for key in [ "id", "ok", "skipped", "reason", "error", "deletedCount", "deletedPvcCount", "protectedCount", "candidateCount", "reclaimedBytes", ] if item.get(key) is not None } command_payload = item.get("command") if isinstance(item.get("command"), dict) else None if command_payload is not None: compact["command"] = { key: command_payload.get(key) for key in ["exitCode", "timedOut", "elapsedMs"] if command_payload.get(key) is not None } results.append(compact) payload = { key: last.get(key) for key in ["ok", "observedAt", "providerId", "unitName", "resultCount", "failedCount", "reclaimedBytes"] if last.get(key) is not None } payload["results"] = results return payload def remote_policy_install_payload(observed_at): rendered = render_remote_policy() try: os.makedirs(rendered["configDir"], mode=0o755, exist_ok=True) os.makedirs(rendered["stateDir"], mode=0o755, exist_ok=True) with open(rendered["scriptPath"], "w", encoding="utf-8") as handle: handle.write(rendered["script"]) os.chmod(rendered["scriptPath"], 0o755) with open(rendered["configPath"], "w", encoding="utf-8") as handle: json.dump(rendered["policyConfig"], handle, sort_keys=True, separators=(",", ":")) handle.write("\n") os.chmod(rendered["configPath"], 0o644) with open(rendered["servicePath"], "w", encoding="utf-8") as handle: handle.write(rendered["service"]) with open(rendered["timerPath"], "w", encoding="utf-8") as handle: handle.write(rendered["timer"]) daemon = command(["systemctl", "daemon-reload"], 30) enable = command(["systemctl", "enable", "--now", "%s.timer" % rendered["unitName"]], 30) status = command(["systemctl", "show", "%s.timer" % rendered["unitName"], "--property=LoadState,ActiveState,SubState,NextElapseUSecRealtime,LastTriggerUSec"], 10) except Exception as exc: return { "ok": False, "action": "gc remote policy install", "providerId": PROVIDER_ID, "dryRun": False, "mutation": True, "observedAt": observed_at, "error": "policy-install-failed", "message": str(exc), } ok = daemon.get("exitCode") == 0 and enable.get("exitCode") == 0 return { "ok": ok, "action": "gc remote policy install", "providerId": PROVIDER_ID, "dryRun": False, "mutation": True, "observedAt": observed_at, "configSource": "config/unidesk-cli.yaml#gc.remote.targets.%s.policyTimer" % PROVIDER_ID, "timer": {key: rendered.get(key) for key in ["unitName", "scriptPath", "configPath", "statePath", "servicePath", "timerPath", "onCalendar", "randomizedDelaySec", "journalTargetBytes", "journalTarget", "tmpMinAgeHours", "includeAptCache", "includeToolCaches", "includeWebObserveArtifacts", "includeK3sImageCache", "includeHostContainerdCache", "includeLocalPathOrphans"]}, "stages": rendered.get("stages"), "systemd": { "daemonReload": bounded(daemon), "enableNow": bounded(enable), "status": bounded(status), }, } def main(): observed_at = now_iso() preflight = cluster_preflight() if ACTION == "policy-plan": emit_json(remote_policy_plan_payload(observed_at), persist_large=False) return 0 if ACTION == "policy-install": emit_json(remote_policy_install_payload(observed_at), persist_large=False) return 0 if ACTION == "policy-status": emit_json(remote_policy_status_payload(observed_at), persist_large=False) return 0 if ACTION == "trend": history_limit = int(OPTIONS.get("historyLimit") or 12) history = read_growth_snapshots(history_limit) emit_json({ "ok": True, "action": "gc remote trend", "providerId": PROVIDER_ID, "dryRun": True, "mutation": False, "observedAt": observed_at, "statePath": growth_snapshot_path(), "historyLimit": history_limit, "trend": growth_trend_payload(history), "points": history if bool(OPTIONS.get("full")) else [compact_growth_point(item) for item in history[-min(len(history), 3):]], "returnedPointCount": min(len(history), 3) if not bool(OPTIONS.get("full")) else len(history), "totalPointCount": len(history), "next": { "snapshot": "bun scripts/cli.ts gc remote %s snapshot --include-hwlab-registry --history-limit %s" % (PROVIDER_ID, history_limit), }, }, persist_large=True) return 0 if ACTION == "snapshot": history_limit = int(OPTIONS.get("historyLimit") or 12) snapshot = collect_growth_snapshot(observed_at, preflight) state_path = growth_snapshot_path() if bool(OPTIONS.get("saveSnapshot", True)): state_path = append_growth_snapshot(snapshot) history = read_growth_snapshots(history_limit) if not bool(OPTIONS.get("saveSnapshot", True)): history = history + [snapshot] trend_payload = growth_trend_payload(history[-history_limit:]) recent_history = history[-min(len(history), 3):] if not bool(OPTIONS.get("full")): trend_payload = compact_trend_payload(trend_payload) recent_history = history[-min(len(history), 1):] snapshot.update({ "statePath": state_path, "historyLimit": history_limit, "saved": bool(OPTIONS.get("saveSnapshot", True)), "trend": trend_payload, "history": { "totalPointCount": len(read_growth_snapshots(1000000)) if bool(OPTIONS.get("saveSnapshot", True)) else len(history), "returnedPointCount": len(recent_history) if bool(OPTIONS.get("full")) else 0, "recentPoints": recent_history if bool(OPTIONS.get("full")) else [], "drillDown": "bun scripts/cli.ts gc remote %s trend --history-limit %s" % (PROVIDER_ID, history_limit), }, }) emit_json(snapshot, persist_large=True) return 0 protected = collect_protected() candidates = collect_candidates(observed_at) visible = visible_items(candidates) if ACTION == "plan": emit_json(plan_payload(observed_at, preflight, protected, candidates, visible), persist_large=True) return 0 if ACTION == "status": emit_json(remote_gc_job_status(), persist_large=False) return 0 if ACTION != "run": emit_json({"ok": False, "error": "unsupported-remote-gc-action", "action": ACTION}, persist_large=False) return 0 if PROVIDER_ID.upper() == "G14" and not preflight.get("ok"): emit_json({ "ok": False, "error": "gc-remote-g14-preflight-failed", "action": "gc remote run", "providerId": PROVIDER_ID, "dryRun": True, "mutation": False, "clusterPreflight": preflight, "plan": plan_payload(observed_at, preflight, protected, candidates, visible), }, persist_large=True) return 0 disk_before = df_snapshot() results = [] for candidate in visible: try: execution = execute(candidate) item = dict(candidate) item.update({"status": execution.get("status") or "succeeded", "reclaimedBytes": execution.get("reclaimedBytes")}) if "commandOutput" in execution: item["commandOutput"] = execution["commandOutput"] results.append(item) except Exception as exc: item = dict(candidate) item.update({"status": "failed", "reclaimedBytes": None, "error": str(exc)}) results.append(item) disk_after = df_snapshot() failed = [item for item in results if item.get("status") == "failed"] returned = returned_results(results) run_summary = summarize(visible, returned, disk_before) run_summary.update({ "plannedCandidateCount": len(visible), "attemptedCount": len(results), "startedCount": len([item for item in results if item.get("status") == "started"]), "succeededCount": len([item for item in results if item.get("status") == "succeeded"]), "failedCount": len(failed), "actualDiskReclaimBytes": (disk_after["availableBytes"] - disk_before["availableBytes"]) if disk_before and disk_after else None, "actualDiskReclaim": fmt_bytes(disk_after["availableBytes"] - disk_before["availableBytes"]) if disk_before and disk_after else None, "targetAfter": target_assessment(disk_after, 0), "resultCount": len(results), "returnedResultCount": len(returned), "omittedResultCount": max(0, len(results) - len(returned)), }) payload = { "ok": len(failed) == 0, "action": "gc remote run", "providerId": PROVIDER_ID, "dryRun": False, "mutation": True, "observedAt": now_iso(), "options": OPTIONS, "diskBefore": disk_before, "diskAfter": disk_after, "clusterPreflight": preflight, "clusterAfter": cluster_preflight(), "summary": run_summary, "results": returned, "protected": protected if bool(OPTIONS.get("full")) else protected[:3], } emit_json(payload, persist_large=True) return 0 if __name__ == "__main__": raise SystemExit(main())