pikasTech-unidesk/scripts/src/gc-remote-runner.py

import base64
import calendar
import fcntl
import json
import os
import re
import signal
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),
        "lock": os.path.join(REMOTE_GC_JOB_DIR, "%s.lock" % job_id),
    }

def process_identity(pid):
    try:
        with open("/proc/%s/stat" % int(pid), "r", encoding="utf-8") as handle:
            text = handle.read().strip()
        close = text.rfind(")")
        fields = text[close + 2:].split() if close >= 0 else []
        return {"state": fields[0], "startTicks": int(fields[19])} if len(fields) >= 20 else None
    except (OSError, ValueError, IndexError):
        return None

def process_start_ticks(pid):
    identity = process_identity(pid)
    return identity.get("startTicks") if identity else None

def acquire_job_lock(paths):
    descriptor = os.open(paths["lock"], os.O_CREAT | os.O_RDWR, 0o600)
    fcntl.flock(descriptor, fcntl.LOCK_EX)
    return descriptor

def release_job_lock(descriptor):
    try:
        fcntl.flock(descriptor, fcntl.LOCK_UN)
    finally:
        os.close(descriptor)

def write_memory_job_state(paths, payload):
    descriptor = acquire_job_lock(paths)
    try:
        write_json_atomic(paths["state"], payload)
    finally:
        release_job_lock(descriptor)

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"]),
        }
    descriptor = None
    try:
        descriptor = acquire_job_lock(paths)
        with open(paths["state"], "r", encoding="utf-8") as handle:
            payload = json.load(handle)
        if payload.get("scope") == "memory-pressure-only" and payload.get("status") == "running":
            worker_pid = safe_int(payload.get("workerPid"))
            expected_start_ticks = safe_int(payload.get("workerStartTicks"))
            actual_identity = process_identity(worker_pid) if worker_pid > 0 else None
            actual_start_ticks = actual_identity.get("startTicks") if actual_identity else None
            worker_alive = actual_identity is not None and actual_identity.get("state") not in set(["Z", "X"]) \
                and actual_start_ticks == expected_start_ticks
            if not worker_alive:
                payload.update({
                    "ok": False,
                    "status": "failed",
                    "outcome": "memory-reclaim-worker-exited-without-terminal-state",
                    "workerAlive": False,
                    "workerObservedState": actual_identity.get("state") if actual_identity else None,
                    "workerObservedStartTicks": actual_start_ticks,
                    "finishedAt": now_iso(),
                })
                write_json_atomic(paths["state"], payload)
            else:
                payload["workerAlive"] = True
    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"]),
        }
    finally:
        if descriptor is not None:
            release_job_lock(descriptor)
    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 <N> --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 <job>" % 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=<redacted>", text_value)
    text_value = re.sub(r"(?i)(--(?:api-key|token|password|secret))\s+\S+", r"\1 <redacted>", 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():
    source = "/proc/meminfo"
    fields = {}
    try:
        with open(source, "r", encoding="utf-8") as handle:
            for line in handle:
                if ":" not in line:
                    continue
                key, raw = line.split(":", 1)
                parts = raw.strip().split()
                if not parts:
                    continue
                fields[key] = safe_int(parts[0]) * 1024
    except OSError as exc:
        return {
            "ok": False,
            "error": "proc-meminfo-read-failed",
            "metric": "MemAvailable",
            "source": source,
            "swapExcluded": True,
            "errorType": type(exc).__name__,
        }
    available_bytes = fields.get("MemAvailable")
    if available_bytes is None:
        return {
            "ok": False,
            "error": "memavailable-missing",
            "metric": "MemAvailable",
            "source": source,
            "swapExcluded": True,
        }
    total_bytes = fields.get("MemTotal")
    memory = {
        "totalBytes": total_bytes,
        "availableBytes": available_bytes,
        "totalHuman": fmt_bytes(total_bytes) if total_bytes is not None else None,
        "availableHuman": fmt_bytes(available_bytes),
    }
    return {
        "ok": True,
        "metric": "MemAvailable",
        "source": source,
        "swapExcluded": True,
        "memory": memory,
    }

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
    live_freshness_timestamp = None
    for source in [heartbeat, manifest]:
        for key in ["updatedAt", "completedAt", "finishedAt", "stoppedAt"]:
            live_freshness_timestamp = iso_or_epoch_to_epoch(source.get(key))
            if live_freshness_timestamp is not None:
                break
        if live_freshness_timestamp 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"])
    age_stale = age_hours >= float(stale_hours)
    stale_signal = (not alive) and age_stale 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,
        "staleLiveSignal": alive and age_stale and live_freshness_timestamp is not None,
        "activeSignal": alive and not (age_stale and live_freshness_timestamp is not None),
        "liveStaleEvidence": "heartbeat-or-manifest-update" if live_freshness_timestamp is not None else None,
        "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 = []
    stale_live_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:
            for child in configured_observe_runs(root):
                try:
                    record = observe_run_record(child, stale_hours)
                except OSError:
                    continue
                row["runCount"] += 1
                run_count += 1
                if record.get("activeSignal"):
                    row["activeSignalCount"] += 1
                    active_rows.append(record)
                if record.get("staleLiveSignal"):
                    stale_live_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),
        "staleLiveSignalCount": len(stale_live_rows),
        "roots": root_rows,
        "activeSignals": active_rows[:int(OPTIONS.get("limit") or 50)],
        "staleSignals": stale_rows[:int(OPTIONS.get("limit") or 50)],
        "staleLiveSignals": stale_live_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 <job>" % 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):
    if bool(OPTIONS.get("memoryPressureOnly")):
        return collect_memory_reclaim_candidates()
    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 summarize_memory_candidates(candidates, returned):
    total_memory = sum(safe_int(item.get("expectedMemoryRssBytes")) for item in candidates)
    returned_memory = sum(safe_int(item.get("expectedMemoryRssBytes")) for item in returned)
    by_kind = {}
    for item in candidates:
        kind = item.get("kind") or "unknown"
        bucket = by_kind.setdefault(kind, {"count": 0, "expectedMemoryRssBytes": 0})
        bucket["count"] += 1
        bucket["expectedMemoryRssBytes"] += safe_int(item.get("expectedMemoryRssBytes"))
        bucket["expectedMemoryRss"] = fmt_bytes(bucket["expectedMemoryRssBytes"])
    return {
        "scope": "memory-pressure-only",
        "candidateCount": len(candidates),
        "returnedCandidateCount": len(returned),
        "expectedMemoryRssBytes": total_memory,
        "expectedMemoryRss": fmt_bytes(total_memory),
        "returnedExpectedMemoryRssBytes": returned_memory,
        "returnedExpectedMemoryRss": fmt_bytes(returned_memory),
        "estimatedDiskBytes": 0,
        "outcome": "memory-reclaim-candidates" if candidates else "no-memory-reclaim",
        "byKind": by_kind,
    }

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.<pid> 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 in set(["browser-orphan-process-tree", "stale-web-observer-process-tree"]):
        return execute_memory_process_candidate(candidate)
    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",
                "estimatedDiskBytes",
                "expectedMemoryRssBytes",
                "reclaimedMemoryBytes",
                "processCount",
                "termSignalCount",
                "killSignalCount",
                "remainingIdentityCount",
                "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):
    if bool(OPTIONS.get("memoryPressureOnly")):
        scan = memory_reclaim_scan_diagnostics()
        summary = summarize_memory_candidates(candidates, visible)
        if scan.get("blocked"):
            summary["outcome"] = "memory-reclaim-scan-blocked"
        return {
            "ok": scan.get("ok") is True,
            "action": "gc remote plan",
            "providerId": PROVIDER_ID,
            "scope": "memory-pressure-only",
            "dryRun": True,
            "mutation": False,
            "observedAt": observed_at,
            "options": OPTIONS,
            "memoryBefore": collect_memory_snapshot(),
            "summary": summary,
            "candidateScan": scan,
            "candidates": visible,
            "protected": {
                "activeObserverIntersection": "excluded" if scan.get("ok") is True else "unknown-protected-all",
                "processIdentity": "pre-term:pid-ppid-sid-startTicks;post-term:pid-startTicks",
                "treeClosure": "unique-root-ppid-sid-closed",
            },
            "policy": {
                "requiresRunConfirm": True,
                "planCommand": "bun scripts/cli.ts gc remote %s plan --memory-pressure-only" % PROVIDER_ID,
                "runCommand": "bun scripts/cli.ts gc remote %s run --confirm --memory-pressure-only" % PROVIDER_ID,
                "candidateKinds": ["browser-orphan-process-tree", "stale-web-observer-process-tree"],
                "neverTouches": ["disk candidates", "apt cache", "journal", "tmp", "container images", "active observer process trees"],
                "termination": "SIGTERM, bounded wait, identity reclose, SIGKILL only same-tree residual",
            },
        }
    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.<pid> 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 update_memory_pressure_job_progress(paths, completed_count, candidate_count, worker_pid, worker_start_ticks):
    descriptor = acquire_job_lock(paths)
    try:
        with open(paths["state"], "r", encoding="utf-8") as handle:
            payload = json.load(handle)
        if payload.get("status") != "running" or safe_int(payload.get("workerPid")) != worker_pid \
                or safe_int(payload.get("workerStartTicks")) != worker_start_ticks:
            raise RuntimeError("memory pressure job state no longer belongs to this worker")
        payload.update({
            "workerHeartbeatAt": now_iso(),
            "progress": {"completedCount": completed_count, "candidateCount": candidate_count},
        })
        write_json_atomic(paths["state"], payload)
    finally:
        release_job_lock(descriptor)

def finish_memory_pressure_job(job_id, candidates, observed_at, worker_pid, worker_start_ticks):
    paths = job_paths(job_id)
    memory_before = collect_memory_snapshot()
    results = []
    update_memory_pressure_job_progress(paths, 0, len(candidates), worker_pid, worker_start_ticks)
    for index, candidate in enumerate(candidates):
        try:
            execution = execute(candidate)
            item = dict(candidate)
            item.update({
                "status": execution.get("status") or "succeeded",
                "reclaimedBytes": 0,
                "reclaimedMemoryBytes": execution.get("reclaimedMemoryBytes"),
                "termSignalCount": execution.get("termSignalCount"),
                "killSignalCount": execution.get("killSignalCount"),
                "remainingIdentityCount": execution.get("remainingIdentityCount"),
                "remainingIdentities": execution.get("remainingIdentities"),
                "lateAllowlistedIdentityCount": execution.get("lateAllowlistedIdentityCount"),
                "memoryBefore": execution.get("memoryBefore"),
                "memoryAfter": execution.get("memoryAfter"),
            })
            results.append(item)
        except Exception as exc:
            item = dict(candidate)
            item.update({"status": "failed", "reclaimedBytes": 0, "reclaimedMemoryBytes": None, "error": str(exc)})
            results.append(item)
        update_memory_pressure_job_progress(paths, index + 1, len(candidates), worker_pid, worker_start_ticks)
    memory_after = collect_memory_snapshot()
    failed = [item for item in results if item.get("status") == "failed"]
    returned = returned_results(results)
    summary = summarize_memory_candidates(candidates, returned)
    summary.update({
        "attemptedCount": len(results),
        "succeededCount": len([item for item in results if item.get("status") == "succeeded"]),
        "failedCount": len(failed),
        "actualMemoryAvailableDeltaBytes": (
            safe_int((memory_after.get("memory") or {}).get("availableBytes"))
            - safe_int((memory_before.get("memory") or {}).get("availableBytes"))
        ) if memory_before.get("ok") and memory_after.get("ok") else None,
    })
    payload = {
        "ok": len(failed) == 0,
        "action": "gc remote run",
        "providerId": PROVIDER_ID,
        "scope": "memory-pressure-only",
        "status": "succeeded" if len(failed) == 0 else "failed",
        "jobId": job_id,
        "statusCommand": status_command(job_id),
        "statePath": paths["state"],
        "dryRun": False,
        "mutation": len(results) > 0,
        "observedAt": observed_at,
        "startedAt": observed_at,
        "finishedAt": now_iso(),
        "workerPid": worker_pid,
        "workerStartTicks": worker_start_ticks,
        "workerAlive": False,
        "workerCompleted": True,
        "workerHeartbeatAt": now_iso(),
        "progress": {"completedCount": len(candidates), "candidateCount": len(candidates)},
        "memoryBefore": memory_before,
        "memoryAfter": memory_after,
        "summary": summary,
        "results": returned,
    }
    write_memory_job_state(paths, payload)

def start_memory_pressure_job(observed_at, candidates):
    job_id = "%s-memory-pressure-%s-%s" % (
        re.sub(r"[^A-Za-z0-9._-]+", "-", PROVIDER_ID.lower()).strip("-") or "provider",
        int(time.time()),
        os.getpid(),
    )
    paths = job_paths(job_id)
    summary = summarize_memory_candidates(candidates, candidates)
    scan = memory_reclaim_scan_diagnostics()
    base = {
        "ok": scan.get("ok") is True,
        "action": "gc remote run",
        "providerId": PROVIDER_ID,
        "scope": "memory-pressure-only",
        "jobId": job_id,
        "statusCommand": status_command(job_id),
        "statePath": paths["state"],
        "dryRun": False,
        "mutation": len(candidates) > 0,
        "observedAt": observed_at,
        "startedAt": observed_at,
        "summary": summary,
        "candidateScan": scan,
    }
    if scan.get("blocked"):
        terminal = dict(base)
        terminal.update({
            "status": "blocked",
            "outcome": "memory-reclaim-scan-blocked",
            "mutation": False,
            "finishedAt": now_iso(),
            "results": [],
        })
        write_json_atomic(paths["state"], terminal)
        return terminal
    if not candidates:
        terminal = dict(base)
        terminal.update({
            "status": "blocked",
            "outcome": "no-memory-reclaim-candidates",
            "mutation": False,
            "finishedAt": now_iso(),
            "results": [],
        })
        write_json_atomic(paths["state"], terminal)
        return terminal
    read_descriptor, write_descriptor = os.pipe()
    try:
        child_pid = os.fork()
    except OSError as exc:
        os.close(read_descriptor)
        os.close(write_descriptor)
        failed = dict(base)
        failed.update({"ok": False, "status": "failed", "outcome": "memory-reclaim-fork-failed", "error": str(exc), "finishedAt": now_iso()})
        write_memory_job_state(paths, failed)
        return failed
    if child_pid == 0:
        try:
            os.close(write_descriptor)
            released = os.read(read_descriptor, 1)
            os.close(read_descriptor)
            if released != b"1":
                os._exit(1)
            os.setsid()
            with open("/dev/null", "rb", buffering=0) as stdin_handle, open(paths["log"], "ab", buffering=0) as log_handle:
                os.dup2(stdin_handle.fileno(), 0)
                os.dup2(log_handle.fileno(), 1)
                os.dup2(log_handle.fileno(), 2)
                finish_memory_pressure_job(job_id, candidates, observed_at, os.getpid(), process_start_ticks(os.getpid()))
        except Exception as exc:
            failed = dict(base)
            failed.update({"ok": False, "status": "failed", "outcome": "memory-reclaim-worker-failed", "workerPid": os.getpid(), "workerStartTicks": process_start_ticks(os.getpid()), "error": str(exc), "finishedAt": now_iso()})
            try:
                write_memory_job_state(paths, failed)
            except Exception:
                pass
        finally:
            os._exit(0)
    os.close(read_descriptor)
    worker_start_ticks = process_start_ticks(child_pid)
    if worker_start_ticks is None:
        os.close(write_descriptor)
        try:
            os.kill(child_pid, signal.SIGKILL)
        except ProcessLookupError:
            pass
        failed = dict(base)
        failed.update({"ok": False, "status": "failed", "outcome": "memory-reclaim-worker-identity-unavailable", "workerPid": child_pid, "finishedAt": now_iso()})
        write_memory_job_state(paths, failed)
        return failed
    running = dict(base)
    running.update({
        "status": "running",
        "outcome": "memory-reclaim-started",
        "workerPid": child_pid,
        "workerStartTicks": worker_start_ticks,
        "workerAlive": True,
        "workerHeartbeatAt": now_iso(),
        "progress": {"completedCount": 0, "candidateCount": len(candidates)},
    })
    try:
        write_memory_job_state(paths, running)
        os.write(write_descriptor, b"1")
    except Exception as exc:
        try:
            os.kill(child_pid, signal.SIGKILL)
        except ProcessLookupError:
            pass
        failed = dict(base)
        failed.update({"ok": False, "status": "failed", "outcome": "memory-reclaim-worker-release-failed", "workerPid": child_pid, "workerStartTicks": worker_start_ticks, "error": str(exc), "finishedAt": now_iso()})
        write_memory_job_state(paths, failed)
        return failed
    finally:
        os.close(write_descriptor)
    return running

def main():
    observed_at = now_iso()
    if ACTION == "memory":
        memory = collect_memory_snapshot()
        emit_json({
            "ok": memory.get("ok") is True,
            "action": "gc remote memory",
            "providerId": PROVIDER_ID,
            "dryRun": True,
            "mutation": False,
            "observedAt": observed_at,
            "metric": memory.get("metric"),
            "source": memory.get("source"),
            "swapExcluded": memory.get("swapExcluded") is True,
            "memory": memory.get("memory"),
            "error": memory.get("error"),
            "valuesRedacted": True,
        }, persist_large=False)
        return 0
    if ACTION == "status" and job_id_or_none():
        emit_json(remote_gc_job_status(), persist_large=False)
        return 0
    if bool(OPTIONS.get("memoryPressureOnly")) and ACTION in set(["plan", "run"]):
        candidates = collect_memory_reclaim_candidates()
        visible = visible_items(candidates)
        if ACTION == "plan":
            emit_json(plan_payload(observed_at, {"skipped": True, "reason": "memory-pressure-only"}, [], candidates, visible), persist_large=True)
        else:
            emit_json(start_memory_pressure_job(observed_at, visible), persist_large=True)
        return 0
    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 = [] if bool(OPTIONS.get("memoryPressureOnly")) else 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())