def memory_distribution_config():
    config = MEMORY_CONFIG.get("attribution")
    if not isinstance(config, dict):
        raise RuntimeError("required YAML memory attribution policy is missing")
    required_integers = {
        "cgroupScanLimit": (1, 1000000),
        "topLevelLimit": (1, 100),
        "systemServiceLimit": (1, 100),
        "podLimit": (1, 100),
        "processLimit": (1, 100),
        "processPssLimit": (1, 50),
        "processSwapLimit": (1, 100),
        "commandTimeoutSeconds": (1, 60),
        "targetMemAvailableBytes": (1, 1 << 50),
        "namespaceCountConcurrency": (1, 32),
    }
    for key, (minimum, maximum) in required_integers.items():
        value = config.get(key)
        if isinstance(value, bool) or not isinstance(value, int) or value < minimum or value > maximum:
            raise RuntimeError("invalid YAML memory attribution integer: %s" % key)
    cgroup_root = config.get("cgroupRoot")
    kubeconfig = config.get("kubeconfigPath")
    resources = config.get("objectCountResources")
    namespaces = config.get("namespaceCountNamespaces")
    if not isinstance(cgroup_root, str) or not cgroup_root.startswith("/") or os.path.normpath(cgroup_root) != cgroup_root:
        raise RuntimeError("invalid YAML memory attribution cgroupRoot")
    if not isinstance(kubeconfig, str) or not kubeconfig.startswith("/") or os.path.normpath(kubeconfig) != kubeconfig:
        raise RuntimeError("invalid YAML memory attribution kubeconfigPath")
    if not isinstance(resources, list) or not resources or any(
        not isinstance(item, str) or not re.match(r"^[a-z0-9.-]{1,100}$", item) for item in resources
    ):
        raise RuntimeError("invalid YAML memory attribution objectCountResources")
    if not isinstance(namespaces, list) or not namespaces or any(
        not isinstance(item, str) or not re.match(r"^[a-z0-9]([-a-z0-9]*[a-z0-9])?$", item) for item in namespaces
    ):
        raise RuntimeError("invalid YAML memory attribution namespaceCountNamespaces")
    return config

def read_memory_distribution_key_values(path, multiplier=1):
    result = {}
    try:
        with open(path, "r", encoding="utf-8") as handle:
            for line in handle:
                parts = line.replace(":", "", 1).split()
                if len(parts) < 2:
                    continue
                try:
                    result[parts[0]] = int(parts[1]) * multiplier
                except ValueError:
                    continue
    except OSError:
        return {}
    return result

def read_memory_distribution_integer(path):
    try:
        with open(path, "r", encoding="utf-8") as handle:
            return int(handle.read().strip())
    except (OSError, ValueError):
        return None

def memory_distribution_cgroup_row(root, path):
    stat = read_memory_distribution_key_values(os.path.join(path, "memory.stat"))
    file_bytes = safe_int(stat.get("file"))
    slab_reclaimable = safe_int(stat.get("slab_reclaimable"))
    return {
        "path": os.path.relpath(path, root),
        "currentBytes": read_memory_distribution_integer(os.path.join(path, "memory.current")),
        "swapCurrentBytes": read_memory_distribution_integer(os.path.join(path, "memory.swap.current")),
        "anonBytes": stat.get("anon"),
        "fileBytes": stat.get("file"),
        "inactiveAnonBytes": stat.get("inactive_anon"),
        "inactiveFileBytes": stat.get("inactive_file"),
        "reclaimableFileSlabUpperBoundBytes": file_bytes + slab_reclaimable,
    }

def memory_distribution_limit(config, key):
    configured = safe_int(config.get(key))
    return min(100, configured * 5) if bool(OPTIONS.get("full")) else configured

def memory_distribution_top(rows, limit):
    return sorted(rows, key=lambda item: safe_int(item.get("currentBytes")), reverse=True)[:limit]

def memory_distribution_command_summary(result):
    return {
        "ok": result.get("exitCode") == 0,
        "exitCode": result.get("exitCode"),
        "timedOut": result.get("timedOut") is True,
        "durationMs": result.get("durationMs"),
    }

def memory_distribution_runtime_maps(config):
    timeout = safe_int(config.get("commandTimeoutSeconds"))
    docker = command(["docker", "ps", "--no-trunc", "--format", "{{.ID}}\t{{.Names}}"], timeout)
    docker_names = {}
    if docker.get("exitCode") == 0:
        for line in docker.get("stdout", "").splitlines():
            parts = line.split("\t", 1)
            if len(parts) == 2 and re.match(r"^[a-f0-9]{64}$", parts[0]):
                docker_names[parts[0]] = parts[1]
    cri = command(["k3s", "crictl", "pods", "-o", "json"], timeout)
    pod_names = {}
    if cri.get("exitCode") == 0:
        try:
            payload = json.loads(cri.get("stdout") or "{}")
            for item in payload.get("items") or []:
                metadata = item.get("metadata") or {}
                uid = str(metadata.get("uid") or "")
                if uid:
                    pod_names[uid] = {
                        "namespace": metadata.get("namespace"),
                        "name": metadata.get("name"),
                        "state": item.get("state"),
                    }
        except (TypeError, ValueError, json.JSONDecodeError):
            pass
    return docker_names, pod_names, {
        "docker": memory_distribution_command_summary(docker),
        "cri": memory_distribution_command_summary(cri),
    }

def collect_memory_distribution_cgroups(config, docker_names, pod_names):
    root = config.get("cgroupRoot")
    scan_limit = safe_int(config.get("cgroupScanLimit"))
    top_level = []
    system_services = []
    pods = []
    scanned = 0
    truncated = False
    pod_pattern = re.compile(r"pod([0-9a-f_]{36})\.slice$")
    docker_pattern = re.compile(r"docker-([a-f0-9]{64})\.scope$")
    for current, directories, files in os.walk(root, topdown=True, followlinks=False):
        directories[:] = sorted(name for name in directories if not os.path.islink(os.path.join(current, name)))
        scanned += 1
        if scanned > scan_limit:
            truncated = True
            break
        if "memory.current" not in files:
            continue
        row = memory_distribution_cgroup_row(root, current)
        parent = os.path.dirname(current)
        if parent == root:
            top_level.append(row)
        if parent == os.path.join(root, "system.slice"):
            match = docker_pattern.search(os.path.basename(current))
            if match:
                row["owner"] = {"kind": "docker", "name": docker_names.get(match.group(1)), "id": match.group(1)}
            system_services.append(row)
        match = pod_pattern.search(os.path.basename(current))
        if match:
            uid = match.group(1).replace("_", "-")
            row["owner"] = {"kind": "kubernetes-pod", "uid": uid, **(pod_names.get(uid) or {})}
            pods.append(row)
            directories[:] = []
    return {
        "scan": {"root": root, "scannedCount": scanned, "limit": scan_limit, "truncated": truncated},
        "topLevel": memory_distribution_top(top_level, memory_distribution_limit(config, "topLevelLimit")),
        "systemServices": memory_distribution_top(system_services, memory_distribution_limit(config, "systemServiceLimit")),
        "pods": memory_distribution_top(pods, memory_distribution_limit(config, "podLimit")),
        "podCgroupCount": len(pods),
    }

def memory_distribution_cgroup_for_pid(pid):
    try:
        with open("/proc/%s/cgroup" % int(pid), "r", encoding="utf-8") as handle:
            for line in handle:
                parts = line.rstrip("\n").split(":", 2)
                if len(parts) == 3 and parts[0] == "0" and parts[1] == "":
                    return parts[2]
    except (OSError, ValueError):
        pass
    return None

def memory_distribution_pss(pid):
    values = read_memory_distribution_key_values("/proc/%s/smaps_rollup" % int(pid), 1024)
    if not values:
        return None
    return {
        "pssBytes": values.get("Pss"),
        "pssAnonBytes": values.get("Pss_Anon"),
        "pssFileBytes": values.get("Pss_File"),
        "swapPssBytes": values.get("SwapPss"),
    }

def memory_distribution_process_swap(pid):
    values = read_memory_distribution_key_values("/proc/%s/status" % int(pid), 1024)
    return values.get("VmSwap")

def memory_distribution_process_owner(cgroup_path, docker_names, pod_names):
    if not isinstance(cgroup_path, str):
        return None
    docker_match = re.search(r"docker-([a-f0-9]{64})\.scope(?:/|$)", cgroup_path)
    if docker_match is not None:
        container_id = docker_match.group(1)
        return {"kind": "docker", "id": container_id, "name": docker_names.get(container_id)}
    match = re.search(r"pod([0-9a-f_]{36})\.slice(?:/|$)", cgroup_path)
    if match is None:
        return None
    uid = match.group(1).replace("_", "-")
    return {"kind": "kubernetes-pod", "uid": uid, **(pod_names.get(uid) or {})}

def collect_memory_distribution_processes(config, docker_names, pod_names):
    result = command(["ps", "-eo", "pid=,ppid=,rss=,stat=,comm=,args="], safe_int(config.get("commandTimeoutSeconds")))
    if result.get("exitCode") != 0:
        return {"ok": False, "command": memory_distribution_command_summary(result), "top": [], "zombies": {}}
    rows = []
    zombie_by_comm = {}
    zombie_by_parent = {}
    for line in result.get("stdout", "").splitlines():
        parts = line.strip().split(None, 5)
        if len(parts) < 5:
            continue
        pid = safe_int(parts[0])
        ppid = safe_int(parts[1])
        rss = safe_int(parts[2]) * 1024
        state = parts[3]
        comm = parts[4]
        args = parts[5] if len(parts) > 5 else comm
        row = {
            "pid": pid,
            "ppid": ppid,
            "comm": comm,
            "state": state,
            "rssBytes": rss,
            "swapBytes": memory_distribution_process_swap(pid),
            "commandPreview": redact_command_preview(args)[:120],
        }
        rows.append(row)
        if state.startswith("Z"):
            zombie_by_comm[comm] = zombie_by_comm.get(comm, 0) + 1
            zombie_by_parent[ppid] = zombie_by_parent.get(ppid, 0) + 1
    by_pid = {item.get("pid"): item for item in rows}
    top = sorted(rows, key=lambda item: safe_int(item.get("rssBytes")), reverse=True)[:memory_distribution_limit(config, "processLimit")]
    top_swap = sorted(rows, key=lambda item: safe_int(item.get("swapBytes")), reverse=True)[:memory_distribution_limit(config, "processSwapLimit")]
    pss_limit = memory_distribution_limit(config, "processPssLimit")
    for index, item in enumerate(top):
        item["cgroupPath"] = memory_distribution_cgroup_for_pid(item.get("pid"))
        item["owner"] = memory_distribution_process_owner(item.get("cgroupPath"), docker_names, pod_names)
        if index < pss_limit:
            item.update(memory_distribution_pss(item.get("pid")) or {})
    for item in top_swap:
        item["cgroupPath"] = memory_distribution_cgroup_for_pid(item.get("pid"))
        item["owner"] = memory_distribution_process_owner(item.get("cgroupPath"), docker_names, pod_names)
    parent_rows = []
    for ppid, count in sorted(zombie_by_parent.items(), key=lambda item: item[1], reverse=True)[:5]:
        parent = by_pid.get(ppid) or {}
        cgroup_path = memory_distribution_cgroup_for_pid(ppid)
        owner = memory_distribution_process_owner(cgroup_path, docker_names, pod_names)
        parent_rows.append({
            "ppid": ppid,
            "parentComm": parent.get("comm"),
            "zombieCount": count,
            "cgroupPath": cgroup_path,
            "owner": owner,
            "classification": "unreaped-child-process-leak",
            "remediation": "fix-child-reaping-and-rollout-owner",
        })
    return {
        "ok": True,
        "command": memory_distribution_command_summary(result),
        "processCount": len(rows),
        "rssTotalBytes": sum(safe_int(item.get("rssBytes")) for item in rows),
        "swapTotalBytes": sum(safe_int(item.get("swapBytes")) for item in rows),
        "top": top,
        "topSwap": top_swap,
        "zombies": {
            "count": sum(zombie_by_comm.values()),
            "byComm": [{"comm": comm, "count": count} for comm, count in sorted(zombie_by_comm.items(), key=lambda item: item[1], reverse=True)[:5]],
            "byParent": parent_rows,
        },
    }

def parse_memory_distribution_pressure(path):
    result = {}
    try:
        with open(path, "r", encoding="utf-8") as handle:
            for line in handle:
                parts = line.split()
                if not parts:
                    continue
                result[parts[0]] = {
                    key: float(value) if key != "total" else safe_int(value)
                    for key, value in (item.split("=", 1) for item in parts[1:] if "=" in item)
                }
    except OSError:
        pass
    return result

def collect_memory_distribution_metrics(config):
    timeout = safe_int(config.get("commandTimeoutSeconds"))
    kubeconfig = config.get("kubeconfigPath")
    result = command(["kubectl", "--kubeconfig", kubeconfig, "get", "--raw", "/metrics"], timeout)
    selected = set(config.get("objectCountResources") or [])
    objects = {}
    runtime = {}
    if result.get("exitCode") == 0:
        object_pattern = re.compile(r'^apiserver_storage_objects\{resource="([^"]+)"\}\s+([0-9.eE+-]+)$')
        runtime_names = set([
            "process_resident_memory_bytes", "process_virtual_memory_bytes", "go_goroutines",
            "go_memstats_heap_alloc_bytes", "go_memstats_heap_inuse_bytes", "go_memstats_heap_sys_bytes",
        ])
        for line in result.get("stdout", "").splitlines():
            match = object_pattern.match(line)
            if match and match.group(1) in selected:
                objects[match.group(1)] = int(float(match.group(2)))
                continue
            parts = line.split()
            if len(parts) == 2 and parts[0] in runtime_names:
                try:
                    runtime[parts[0]] = float(parts[1])
                except ValueError:
                    pass
    return {"command": memory_distribution_command_summary(result), "objects": objects, "runtime": runtime}

def memory_distribution_namespace_resource_path(resource, namespace):
    encoded_namespace = urllib.parse.quote(namespace, safe="")
    paths = {
        "pods": "/api/v1/namespaces/%s/pods?limit=1" % encoded_namespace,
        "events": "/api/v1/namespaces/%s/events?limit=1" % encoded_namespace,
        "secrets": "/api/v1/namespaces/%s/secrets?limit=1" % encoded_namespace,
        "persistentvolumeclaims": "/api/v1/namespaces/%s/persistentvolumeclaims?limit=1" % encoded_namespace,
        "pipelineruns.tekton.dev": "/apis/tekton.dev/v1/namespaces/%s/pipelineruns?limit=1" % encoded_namespace,
        "taskruns.tekton.dev": "/apis/tekton.dev/v1/namespaces/%s/taskruns?limit=1" % encoded_namespace,
    }
    return paths.get(resource)

def memory_distribution_namespace_count(config, namespace, resource):
    path = memory_distribution_namespace_resource_path(resource, namespace)
    if path is None:
        return namespace, resource, None, "unsupported-resource"
    result = command([
        "kubectl", "--kubeconfig", config.get("kubeconfigPath"), "get", "--raw", path,
    ], safe_int(config.get("commandTimeoutSeconds")))
    if result.get("exitCode") != 0:
        return namespace, resource, None, "read-failed"
    try:
        payload = json.loads(result.get("stdout") or "{}")
        metadata = payload.get("metadata") or {}
        count = len(payload.get("items") or []) + safe_int(metadata.get("remainingItemCount"))
        return namespace, resource, count, None
    except (TypeError, ValueError, json.JSONDecodeError):
        return namespace, resource, None, "parse-failed"

def collect_memory_distribution_namespace_counts(config):
    import concurrent.futures

    namespaces = list(config.get("namespaceCountNamespaces") or [])
    resources = [
        item for item in (config.get("objectCountResources") or [])
        if memory_distribution_namespace_resource_path(item, namespaces[0]) is not None
    ]
    jobs = [(namespace, resource) for namespace in namespaces for resource in resources]
    rows = {namespace: {"namespace": namespace} for namespace in namespaces}
    errors = []
    with concurrent.futures.ThreadPoolExecutor(max_workers=safe_int(config.get("namespaceCountConcurrency"))) as pool:
        futures = [pool.submit(memory_distribution_namespace_count, config, namespace, resource) for namespace, resource in jobs]
        for future in concurrent.futures.as_completed(futures):
            namespace, resource, count, error = future.result()
            if error is None:
                rows[namespace][resource] = count
            else:
                errors.append({"namespace": namespace, "resource": resource, "error": error})
    ordered = sorted(
        rows.values(),
        key=lambda item: sum(safe_int(value) for key, value in item.items() if key != "namespace"),
        reverse=True,
    )
    return {"rows": ordered, "errors": errors, "complete": not errors}

def compact_memory_distribution_cgroup(item):
    compact = {
        key: item.get(key)
        for key in ["currentBytes", "swapCurrentBytes", "anonBytes"]
        if item.get(key) is not None
    }
    owner = item.get("owner") if isinstance(item.get("owner"), dict) else None
    if owner is not None:
        compact["owner"] = {
            key: owner.get(key)
            for key in ["kind", "namespace", "name"]
            if owner.get(key) is not None
        }
    else:
        compact["path"] = item.get("path")
    return compact

def compact_memory_distribution_process(item):
    compact = {
        key: item.get(key)
        for key in ["pid", "comm", "rssBytes", "swapBytes", "pssBytes"]
        if item.get(key) is not None
    }
    owner = item.get("owner") if isinstance(item.get("owner"), dict) else None
    if owner is not None:
        compact["owner"] = {
            key: owner.get(key)
            for key in ["kind", "namespace", "name"]
            if owner.get(key) is not None
        }
    else:
        compact["cgroupPath"] = item.get("cgroupPath")
    return compact

def compact_memory_distribution_pressure(pressure):
    return {
        level: {
            key: values.get(key)
            for key in ["avg10", "avg60"]
            if values.get(key) is not None
        }
        for level, values in pressure.items()
        if isinstance(values, dict)
    }

def compact_memory_distribution_zombies(zombies):
    by_parent = []
    for item in zombies.get("byParent") or []:
        compact = {
            key: item.get(key)
            for key in ["ppid", "parentComm", "zombieCount", "classification", "remediation"]
            if item.get(key) is not None
        }
        owner = item.get("owner") if isinstance(item.get("owner"), dict) else None
        if owner is not None:
            compact["owner"] = {
                key: owner.get(key)
                for key in ["kind", "namespace", "name"]
                if owner.get(key) is not None
            }
        by_parent.append(compact)
    return {
        "count": zombies.get("count"),
        "byComm": (zombies.get("byComm") or [])[:3],
        "byParent": by_parent[:3],
    }

def collect_memory_distribution(observed_at):
    try:
        config = memory_distribution_config()
    except RuntimeError as exc:
        return {"ok": False, "action": "gc remote memory-distribution", "providerId": PROVIDER_ID, "dryRun": True, "mutation": False, "error": str(exc)}
    started = time.time()
    docker_names, pod_names, runtime_commands = memory_distribution_runtime_maps(config)
    host_snapshot = collect_memory_snapshot()
    meminfo = read_memory_distribution_key_values("/proc/meminfo", 1024)
    cgroups = collect_memory_distribution_cgroups(config, docker_names, pod_names)
    processes = collect_memory_distribution_processes(config, docker_names, pod_names)
    metrics = collect_memory_distribution_metrics(config)
    namespace_counts = collect_memory_distribution_namespace_counts(config)
    available = safe_int((host_snapshot.get("memory") or {}).get("availableBytes"))
    target = safe_int(config.get("targetMemAvailableBytes"))
    cached = safe_int(meminfo.get("Cached")) + safe_int(meminfo.get("Buffers"))
    slab_reclaimable = safe_int(meminfo.get("SReclaimable"))
    payload = {
        "ok": cgroups.get("scan", {}).get("truncated") is not True,
        "action": "gc remote memory-distribution",
        "providerId": PROVIDER_ID,
        "dryRun": True,
        "mutation": False,
        "observedAt": observed_at,
        "durationMs": int((time.time() - started) * 1000),
        "summary": {
            "metric": "MemAvailable",
            "source": "/proc/meminfo",
            "swapExcludedFromAvailability": True,
            "availableBytes": available,
            "targetAvailableBytes": target,
            "targetGapBytes": max(0, target - available + 1),
            "anonPagesBytes": meminfo.get("AnonPages"),
            "cachedAndBuffersBytes": cached,
            "slabBytes": meminfo.get("Slab"),
            "reclaimableFileSlabUpperBoundBytes": cached + slab_reclaimable,
            "swapFreeBytes": safe_int((host_snapshot.get("swap") or {}).get("freeBytes")),
            "memoryPsiFullAvg60": (parse_memory_distribution_pressure("/proc/pressure/memory").get("full") or {}).get("avg60"),
        },
        "host": {
            "snapshot": host_snapshot,
            "pages": {key: meminfo.get(key) for key in [
                "MemFree", "Buffers", "Cached", "Active", "Inactive", "Active(anon)", "Inactive(anon)",
                "Active(file)", "Inactive(file)", "AnonPages", "Mapped", "Shmem", "Slab", "SReclaimable",
                "SUnreclaim", "KernelStack", "PageTables", "SwapCached",
            ]},
            "pressure": {resource: parse_memory_distribution_pressure("/proc/pressure/%s" % resource) for resource in ["memory", "io", "cpu"]},
        },
        "cgroups": cgroups,
        "processes": processes,
        "kubernetes": {**metrics, "namespaceCounts": namespace_counts},
        "collection": {
            "configSource": "config/unidesk-cli.yaml#gc.remote.targets.%s.memoryPressure.attribution" % PROVIDER_ID,
            "runtimeMaps": runtime_commands,
            "podRuntimeMapCount": len(pod_names),
            "dockerRuntimeMapCount": len(docker_names),
            "full": bool(OPTIONS.get("full")),
        },
        "policy": {
            "analysisOnly": True,
            "next": "Use this distribution before designing or running memory-pressure GC candidates.",
        },
    }
    if bool(OPTIONS.get("full")):
        return payload
    payload["host"] = {
        "pressure": {"memory": compact_memory_distribution_pressure(parse_memory_distribution_pressure("/proc/pressure/memory"))},
    }
    payload["cgroups"] = {
        "scan": cgroups.get("scan"),
        "podCgroupCount": cgroups.get("podCgroupCount"),
        "topLevel": [compact_memory_distribution_cgroup(item) for item in cgroups.get("topLevel") or []],
        "systemServices": [compact_memory_distribution_cgroup(item) for item in cgroups.get("systemServices") or []],
        "pods": [compact_memory_distribution_cgroup(item) for item in cgroups.get("pods") or []],
    }
    payload["processes"] = {
        "ok": processes.get("ok"),
        "processCount": processes.get("processCount"),
        "rssTotalBytes": processes.get("rssTotalBytes"),
        "swapTotalBytes": processes.get("swapTotalBytes"),
        "top": [compact_memory_distribution_process(item) for item in processes.get("top") or []],
        "topSwap": [compact_memory_distribution_process(item) for item in processes.get("topSwap") or []],
        "zombies": compact_memory_distribution_zombies(processes.get("zombies") or {}),
    }
    payload["kubernetes"] = {
        "objects": metrics.get("objects"),
        "namespaceCounts": namespace_counts,
    }
    if (metrics.get("command", {}).get("ok") is not True):
        payload["kubernetes"]["commandError"] = metrics.get("command")
    payload["collection"] = {
        "configSource": payload.get("collection", {}).get("configSource"),
        "podRuntimeMapCount": len(pod_names),
        "dockerRuntimeMapCount": len(docker_names),
        "runtimeMapErrors": {
            key: value for key, value in runtime_commands.items()
            if value.get("ok") is not True
        },
        "full": False,
    }
    return payload