def k3s_crictl_base():
    endpoint = str(CONTAINERD_CONFIG.get("runtimeEndpoint") or "unix:///run/k3s/containerd/containerd.sock")
    return ["crictl", "--runtime-endpoint", endpoint]

def shell_single_quote(value):
    return "'" + str(value).replace("'", "'\"'\"'") + "'"

def k3s_crictl_json(args, timeout=30):
    result = command(k3s_crictl_base() + args + ["-o", "json"], timeout)
    if result["exitCode"] != 0:
        return None, result
    try:
        return json.loads(result["stdout"] or "{}"), result
    except Exception:
        return None, result

def ci_activity_snapshot_for_prune():
    namespaces = config_list(CONTAINERD_CONFIG, "ciNamespaces", ["hwlab-ci", "agentrun-ci"])
    active = []
    commands = []
    for namespace in namespaces:
        result = command(["sh", "-lc", "KUBECONFIG=/etc/rancher/k3s/k3s.yaml kubectl get pipelinerun,taskrun,job -n %s --no-headers 2>/dev/null | awk '$2 != \"True\" && $2 != \"False\" && $2 != \"Complete\" && $2 != \"Failed\" {print}' | head -20" % shell_single_quote(namespace)], 15)
        commands.append({"namespace": namespace, "command": bounded(result)})
        for line in (result.get("stdout") or "").splitlines():
            if line.strip():
                active.append({"namespace": namespace, "line": line.strip()})
    return {"ok": True, "activeCount": len(active), "activePreview": active[:20], "commands": commands}

def compact_ci_activity(activity):
    return {
        "ok": activity.get("ok"),
        "activeCount": activity.get("activeCount"),
        "activePreview": activity.get("activePreview") or [],
    }

def compact_image_ref(ref):
    ref = str(ref or "")
    return ref if len(ref) <= 120 else ref[:117] + "..."

def k3s_cri_image_rows():
    images, image_cmd = k3s_crictl_json(["images"], 45)
    containers, container_cmd = k3s_crictl_json(["ps", "-a"], 30)
    if images is None:
        return None, {"ok": False, "reason": "crictl-images-failed", "command": bounded(image_cmd)}
    if containers is None:
        return None, {"ok": False, "reason": "crictl-ps-failed", "command": bounded(container_cmd)}
    used = set()
    for container in containers.get("containers") or []:
        for key in ["imageRef", "image", "imageId"]:
            value = container.get(key)
            if isinstance(value, str) and value:
                used.add(value)
        image = container.get("image") or {}
        if isinstance(image, dict):
            for key in ["image", "annotations", "userSpecifiedImage"]:
                value = image.get(key)
                if isinstance(value, str) and value:
                    used.add(value)
    rows = []
    for image in images.get("images") or []:
        refs = []
        for key in ["repoTags", "repoDigests"]:
            value = image.get(key)
            if isinstance(value, list):
                refs.extend([str(item) for item in value if item])
        image_id = str(image.get("id") or "")
        pinned = bool(image.get("pinned"))
        size = safe_int(image.get("size_") or image.get("size") or 0)
        in_use = pinned or image_id in used or any(ref in used for ref in refs)
        rows.append({"id": image_id, "refs": refs, "sizeBytes": size, "inUse": in_use, "pinned": pinned})
    return rows, {"ok": True, "imageCommand": bounded(image_cmd), "containerCommand": bounded(container_cmd)}

def k3s_image_cache_candidate():
    if not config_bool(CONTAINERD_CONFIG, "enabled", False):
        return {
            "id": "k3s-cri-image-prune:disabled",
            "kind": "k3s-cri-image-prune-disabled",
            "risk": "blocked",
            "description": "K3s CRI image prune is disabled in YAML",
            "estimatedReclaimBytes": 0,
            "configSource": "config/unidesk-cli.yaml#gc.remote.targets.%s.containerdImageCache.enabled" % PROVIDER_ID,
        }
    activity = ci_activity_snapshot_for_prune()
    if int(activity.get("activeCount") or 0) > 0:
        return {
            "id": "k3s-cri-image-prune:ci-active",
            "kind": "k3s-cri-image-prune-blocked",
            "risk": "blocked",
            "description": "K3s CRI image prune is blocked while CI workloads are active",
            "estimatedReclaimBytes": 0,
            "ciActivity": compact_ci_activity(activity),
        }
    rows, meta = k3s_cri_image_rows()
    if rows is None:
        return {
            "id": "k3s-cri-image-prune:unavailable",
            "kind": "k3s-cri-image-prune-unavailable",
            "risk": "blocked",
            "description": "K3s CRI image list is unavailable",
            "estimatedReclaimBytes": 0,
            "diagnostic": meta,
        }
    unused = [row for row in rows if not row.get("inUse")]
    estimated = sum(safe_int(row.get("sizeBytes")) for row in unused)
    if estimated <= 0:
        return None
    return {
        "id": "k3s-cri-image-prune:unused",
        "kind": "k3s-cri-image-prune",
        "risk": "medium",
        "description": "Prune unused k3s CRI images through crictl rmi --prune; no containerd paths are deleted directly",
        "sizeBytes": estimated,
        "estimatedReclaimBytes": estimated,
        "imageCount": len(rows),
        "unusedImageCount": len(unused),
        "unusedPreview": [{"id": row.get("id"), "refs": [compact_image_ref(ref) for ref in (row.get("refs") or [])[:2]], "sizeBytes": row.get("sizeBytes")} for row in unused[:3]],
        "ciActivity": compact_ci_activity(activity),
        "action": {"command": k3s_crictl_base() + ["rmi", "--prune"], "mode": "cri-unused-images-only"},
    }

def execute_k3s_image_cache_prune():
    activity = ci_activity_snapshot_for_prune()
    if int(activity.get("activeCount") or 0) > 0:
        raise RuntimeError("refusing k3s image prune while CI workloads are active")
    before = du_size("/var/lib/rancher/k3s/agent/containerd", 45) or 0
    result = command(k3s_crictl_base() + ["rmi", "--prune"], 300)
    if result["exitCode"] != 0:
        raise RuntimeError((result["stderr"] or result["stdout"] or "crictl rmi --prune failed").strip())
    after = du_size("/var/lib/rancher/k3s/agent/containerd", 45) or 0
    return {"reclaimedBytes": max(0, before - after), "commandOutput": bounded(result), "ciActivity": compact_ci_activity(activity)}

def host_ctr_base(namespace=None):
    address = config_str(HOST_CONTAINERD_CONFIG, "address", "")
    args = ["ctr"]
    if address:
        args.extend(["--address", address])
    if namespace:
        args.extend(["-n", namespace])
    return args

def host_ctr(args, timeout=30, namespace=None):
    return command(host_ctr_base(namespace) + args, timeout)

def host_containerd_namespaces():
    configured = config_list(HOST_CONTAINERD_CONFIG, "namespaces", [])
    if configured:
        return configured, {"source": "yaml", "command": None}
    result = host_ctr(["namespaces", "list", "-q"], 20)
    if result["exitCode"] != 0:
        return [], {"source": "ctr", "command": bounded(result), "error": "ctr-namespaces-failed"}
    return [line.strip() for line in (result.get("stdout") or "").splitlines() if line.strip()], {"source": "ctr", "command": bounded(result)}

def host_containerd_activity():
    if not config_bool(HOST_CONTAINERD_CONFIG, "enabled", False):
        return {"ok": False, "reason": "host-containerd-cache-disabled", "activeCount": 0}
    root = config_str(HOST_CONTAINERD_CONFIG, "root", "")
    if not root or not os.path.isdir(root):
        return {"ok": False, "reason": "host-containerd-root-unavailable", "root": root, "activeCount": 0}
    namespaces, namespace_meta = host_containerd_namespaces()
    active = []
    commands = []
    for namespace in namespaces:
        task_result = host_ctr(["tasks", "list", "-q"], 20, namespace)
        container_result = host_ctr(["containers", "list", "-q"], 20, namespace)
        image_result = host_ctr(["images", "list", "-q"], 20, namespace)
        lease_result = host_ctr(["leases", "list", "-q"], 20, namespace)
        snapshot_result = host_ctr(["snapshots", "list"], 20, namespace)
        content_result = host_ctr(["content", "list"], 20, namespace)
        snapshot_lines = table_data_lines(snapshot_result.get("stdout") or "", "KEY")
        content_lines = table_data_lines(content_result.get("stdout") or "", "DIGEST")
        commands.append({
            "namespace": namespace,
            "tasks": bounded(task_result),
            "containers": bounded(container_result),
            "images": bounded(image_result),
            "leases": bounded(lease_result),
            "snapshots": bounded(snapshot_result),
            "content": bounded(content_result),
        })
        for kind, result in [("task", task_result), ("container", container_result), ("lease", lease_result)]:
            if result["exitCode"] != 0:
                active.append({"namespace": namespace, "kind": kind, "state": "unknown", "reason": "ctr-list-failed"})
                continue
            for line in (result.get("stdout") or "").splitlines():
                if line.strip():
                    active.append({"namespace": namespace, "kind": kind, "name": line.strip()})
        if snapshot_result["exitCode"] != 0:
            active.append({"namespace": namespace, "kind": "snapshot", "state": "unknown", "reason": "ctr-list-failed"})
        for line in snapshot_lines:
            active.append({"namespace": namespace, "kind": "snapshot", "name": line.split()[0] if line.split() else line})
        if content_result["exitCode"] != 0:
            active.append({"namespace": namespace, "kind": "content", "state": "unknown", "reason": "ctr-list-failed"})
        for line in content_lines:
            active.append({"namespace": namespace, "kind": "content", "name": line.split()[0] if line.split() else line})
    return {
        "ok": True,
        "root": root,
        "namespaces": namespaces,
        "namespaceMeta": namespace_meta,
        "activeCount": len(active),
        "activePreview": active[:20],
        "commands": commands,
    }

def compact_host_containerd_activity(activity):
    return {
        "ok": activity.get("ok"),
        "reason": activity.get("reason"),
        "root": activity.get("root"),
        "namespaces": activity.get("namespaces"),
        "activeCount": activity.get("activeCount"),
        "activePreview": activity.get("activePreview") or [],
    }

def table_data_lines(stdout, header_prefix):
    lines = [line.strip() for line in str(stdout or "").splitlines() if line.strip()]
    return [line for line in lines if not line.startswith(header_prefix)]

def host_containerd_orphan_config():
    value = HOST_CONTAINERD_CONFIG.get("orphanCleanup") if isinstance(HOST_CONTAINERD_CONFIG, dict) else None
    return value if isinstance(value, dict) else {}

def direct_child_paths(root, predicate):
    if not root or not os.path.isdir(root) or os.path.islink(root):
        return []
    rows = []
    for name in sorted(os.listdir(root)):
        path = os.path.realpath(os.path.abspath(os.path.join(root, name)))
        if os.path.dirname(path) != os.path.realpath(os.path.abspath(root)):
            continue
        if not predicate(name, path):
            continue
        rows.append({"name": name, "path": path, "estimatedReclaimBytes": du_size(path, 10) or path_size(path)})
    return rows

def host_containerd_orphan_rows(activity):
    cfg = host_containerd_orphan_config()
    if not config_bool(cfg, "enabled", False):
        return [], {"ok": False, "reason": "host-containerd-orphan-cleanup-disabled"}
    if not activity.get("ok") or int(activity.get("activeCount") or 0) > 0:
        return [], {"ok": False, "reason": "host-containerd-metadata-not-empty", "activity": compact_host_containerd_activity(activity)}
    overlay_root = os.path.realpath(os.path.abspath(config_str(cfg, "overlaySnapshotsRoot", "")))
    content_root = os.path.realpath(os.path.abspath(config_str(cfg, "contentBlobRoot", "")))
    root = os.path.realpath(os.path.abspath(config_str(HOST_CONTAINERD_CONFIG, "root", "")))
    if not root or not overlay_root.startswith(root.rstrip("/") + "/") or not content_root.startswith(root.rstrip("/") + "/"):
        return [], {"ok": False, "reason": "host-containerd-orphan-root-outside-containerd-root", "root": root}
    open_roots = []
    for candidate_root in [overlay_root, content_root]:
        if os.path.exists(candidate_root) and path_has_open_fd(candidate_root):
            open_roots.append(candidate_root)
    if open_roots:
        return [], {"ok": False, "reason": "host-containerd-orphan-root-open-fd", "openRoots": open_roots}
    overlay_rows = direct_child_paths(overlay_root, lambda name, path: os.path.isdir(path) and not os.path.islink(path) and re.match(r"^[0-9]+$", name) is not None)
    content_rows = direct_child_paths(content_root, lambda name, path: os.path.isfile(path) and not os.path.islink(path) and re.match(r"^[0-9a-f]{64}$", name) is not None)
    safe_rows = []
    for kind, rows in [("overlay-snapshot-dir", overlay_rows), ("content-blob-file", content_rows)]:
        for row in rows:
            safe_rows.append({**row, "kind": kind})
    safe_rows.sort(key=lambda item: safe_int(item.get("estimatedReclaimBytes")), reverse=True)
    return safe_rows, {
        "ok": True,
        "root": root,
        "overlaySnapshotsRoot": overlay_root,
        "contentBlobRoot": content_root,
        "overlayCandidateCount": len(overlay_rows),
        "contentCandidateCount": len(content_rows),
        "protectedCount": 0,
        "protectedPreview": [],
    }

def host_containerd_orphan_candidate(activity):
    rows, meta = host_containerd_orphan_rows(activity)
    if not meta.get("ok"):
        return None
    limit = int(OPTIONS.get("limit") or 50)
    selected = rows[:limit]
    estimated = sum(safe_int(row.get("estimatedReclaimBytes")) for row in selected)
    if estimated <= 0:
        return None
    return {
        "id": "host-containerd-orphan-state:delete",
        "kind": "host-containerd-orphan-state-delete",
        "risk": "medium",
        "description": "Delete YAML-allowlisted host containerd orphan snapshot/content files only when ctr metadata has no tasks, containers, leases, images, snapshots or content",
        "path": meta.get("root"),
        "sizeBytes": estimated,
        "estimatedReclaimBytes": estimated,
        "orphanCount": len(rows),
        "selectedOrphanCount": len(selected),
        "overlayCandidateCount": meta.get("overlayCandidateCount"),
        "contentCandidateCount": meta.get("contentCandidateCount"),
        "protectedCount": meta.get("protectedCount"),
        "selectedPreview": [{"kind": row.get("kind"), "name": row.get("name"), "estimatedReclaimBytes": row.get("estimatedReclaimBytes")} for row in selected[:8]],
        "protectedPreview": meta.get("protectedPreview"),
        "action": {"op": "remove-yaml-allowlisted-host-containerd-orphans", "limit": limit},
    }

def host_containerd_cache_candidate():
    activity = host_containerd_activity()
    if not activity.get("ok"):
        return {
            "id": "host-containerd-cache:unavailable",
            "kind": "host-containerd-cache-unavailable",
            "risk": "blocked",
            "description": "Host containerd cache cleanup is disabled or unavailable by YAML",
            "estimatedReclaimBytes": 0,
            "diagnostic": compact_host_containerd_activity(activity),
        }
    if int(activity.get("activeCount") or 0) > 0:
        return {
            "id": "host-containerd-cache:active",
            "kind": "host-containerd-cache-blocked",
            "risk": "blocked",
            "description": "Host containerd cache prune is blocked while host containerd tasks or containers exist",
            "estimatedReclaimBytes": 0,
            "activity": compact_host_containerd_activity(activity),
        }
    orphan = host_containerd_orphan_candidate(activity)
    if orphan:
        return orphan
    root = activity.get("root") or ""
    size = du_size(root, 45) or 0
    if size <= 0:
        return None
    return {
        "id": "host-containerd-cache:prune-unused",
        "kind": "host-containerd-cache-prune",
        "risk": "medium",
        "description": "Prune host containerd images in YAML-selected namespaces only when no host containerd tasks or containers exist",
        "path": root,
        "sizeBytes": size,
        "estimatedReclaimBytes": size,
        "activity": compact_host_containerd_activity(activity),
        "action": {"command": "ctr images prune --all per namespace", "mode": "host-containerd-unused-images-only"},
    }

def execute_host_containerd_cache_prune():
    activity = host_containerd_activity()
    if not activity.get("ok"):
        raise RuntimeError("host containerd cache cleanup unavailable: %s" % activity.get("reason"))
    if int(activity.get("activeCount") or 0) > 0:
        raise RuntimeError("refusing host containerd prune while tasks or containers exist")
    root = activity.get("root") or ""
    before = du_size(root, 45) or 0
    results = []
    for namespace in activity.get("namespaces") or []:
        result = host_ctr(["images", "prune", "--all"], 300, namespace)
        results.append({"namespace": namespace, "imagesPrune": bounded(result)})
        if result["exitCode"] != 0:
            raise RuntimeError("host containerd image prune failed in namespace %s: %s" % (namespace, (result.get("stderr") or result.get("stdout") or "").strip()))
    after = du_size(root, 45) or 0
    return {
        "reclaimedBytes": max(0, before - after),
        "activity": compact_host_containerd_activity(activity),
        "commandResults": results[:8],
    }

def execute_host_containerd_orphan_cleanup():
    activity = host_containerd_activity()
    rows, meta = host_containerd_orphan_rows(activity)
    if not meta.get("ok"):
        raise RuntimeError("host containerd orphan cleanup unavailable: %s" % meta.get("reason"))
    for root_path in [meta.get("overlaySnapshotsRoot"), meta.get("contentBlobRoot")]:
        if root_path and os.path.exists(root_path) and path_has_open_fd(root_path):
            raise RuntimeError("refusing host containerd orphan cleanup with open fd/cwd under root: %s" % root_path)
    limit = int(OPTIONS.get("limit") or 50)
    selected = rows[:limit]
    reclaimed = 0
    deleted = []
    for row in selected:
        path = row.get("path")
        before = du_size(path, 10) or path_size(path)
        if row.get("kind") == "overlay-snapshot-dir":
            shutil.rmtree(path, ignore_errors=True)
        elif row.get("kind") == "content-blob-file":
            os.unlink(path)
        else:
            raise RuntimeError("unsupported host containerd orphan kind: %s" % row.get("kind"))
        reclaimed += before
        deleted.append({"kind": row.get("kind"), "name": row.get("name"), "reclaimedBytes": before})
    return {
        "reclaimedBytes": reclaimed,
        "deletedOrphanCount": len(deleted),
        "deletedPreview": deleted[:12],
        "root": meta.get("root"),
    }