Merge pull request #2267 from pikasTech/fix/1710-nc01-zombie-reaper

fix: 收敛 NC01 内存与 Tekton 留存

.agents/skills/unidesk-gc/SKILL.md

@@ -120,6 +120,32 @@ rm -rf -- /tmp/unidesk-cli-output
 - provider host 统一使用 `bun scripts/cli.ts gc remote <providerId> ...`。
 - 远端长任务必须异步执行，并用 `status --job-id` 短查询，不得维持长 SSH 会话。
 
+### 内存分布与 Kubernetes 留存
+
+先用紧凑 overview 归因真实内存压力：
+
+```bash
+bun scripts/cli.ts gc remote <providerId> memory-distribution
+```
+
+- 默认输出物理内存、PSS、cgroup、zombie 和 Kubernetes 对象分布；
+- swap 单独披露，不能计入 WebProbe 启动资格；
+- 需要完整明细时才使用 `--full`。
+
+Kubernetes 历史对象手动回收固定使用：
+
+```bash
+bun scripts/cli.ts gc remote <providerId> retention plan --limit 500
+bun scripts/cli.ts gc remote <providerId> retention run --confirm --limit 500
+bun scripts/cli.ts gc remote <providerId> status --job-id <id>
+```
+
+- 筛选、保留窗口、分组、批量和级联策略只来自目标节点的 `kubernetesObjectRetention` YAML；
+- 手动入口与 `policyTimer.includeKubernetesObjectRetention` 自动阶段必须复用同一筛选器；
+- 活跃、过新、未知分组、时间不可解析、扫描截断和 Table 形状漂移全部保护；
+- 只删除 YAML 指定的根对象，通过 ownerRef 级联，不直接删除下游 TaskRun、Pod、Secret 或 Event；
+- 自动回收由 `gc remote <providerId> policy plan|install --confirm|status` 管理，周期和单批上限由 YAML 控制。
+
 ## WebProbe 低内存人工处置
 
 - 唯一资格指标：

config/unidesk-cli.yaml

@@ -160,7 +160,33 @@ gc:
           observeRunDepth: 6
           observeScanLimit: 2000
           staleRunMaxAgeHours: 1
-          processScanLimit: 16384
+          processScanLimit: 32768
+          attribution:
+            cgroupRoot: /sys/fs/cgroup
+            cgroupScanLimit: 32768
+            topLevelLimit: 4
+            systemServiceLimit: 4
+            podLimit: 4
+            processLimit: 4
+            processPssLimit: 4
+            commandTimeoutSeconds: 20
+            kubeconfigPath: /etc/rancher/k3s/k3s.yaml
+            targetMemAvailableBytes: 4294967296
+            objectCountResources:
+              - pods
+              - pipelineruns.tekton.dev
+              - taskruns.tekton.dev
+              - events
+              - secrets
+              - persistentvolumeclaims
+              - persistentvolumes
+            namespaceCountNamespaces:
+              - devops-infra
+              - agentrun-ci
+              - hwlab-ci
+              - pikaoa-ci
+              - selfmedia-ci
+            namespaceCountConcurrency: 8
           processStaleMinAgeSeconds: 3600
           maxMemoryReclaimCandidates: 20
           terminationGraceSeconds: 5
@@ -181,6 +207,59 @@ gc:
             - /usr/bin/chromium-browser
             - /opt/google/chrome/
             - /headless_shell
+          cgroupReclaim:
+            enabled: false
+        kubernetesObjectRetention:
+          enabled: true
+          configId: devops-infra-tekton-history
+          kubeconfigPath: /etc/rancher/k3s/k3s.yaml
+          namespace: devops-infra
+          resource: pipelineruns.tekton.dev
+          listMode: server-table
+          countPath: /apis/tekton.dev/v1/namespaces/{namespace}/pipelineruns
+          terminalConditionType: Succeeded
+          terminalStatuses:
+            - "True"
+            - "False"
+          minAgeHours: 24
+          keepLatestPerGroup: 3
+          manualMaxDeletePerRun: 500
+          automaticMaxDeletePerRun: 100
+          previewLimit: 10
+          scanLimit: 5000
+          requestTimeoutSeconds: 45
+          deleteBatchSize: 50
+          deleteTimeoutSeconds: 60
+          postDeleteSettleSeconds: 5
+          propagationPolicy: Background
+          groupSelectors:
+            - id: outer-pac-event
+              groupByLabels:
+                - pipelinesascode.tekton.dev/repository
+                - pipelinesascode.tekton.dev/original-prname
+            - id: inner-deterministic-publish
+              groupByLabels:
+                - unidesk.ai/node
+                - unidesk.ai/lane
+                - unidesk.ai/spec-ref
+        policyTimer:
+          enabled: true
+          name: unidesk-devops-infra-retention
+          onCalendar: "*:0/15"
+          randomizedDelaySec: 2min
+          stateDir: /var/lib/unidesk-gc
+          configDir: /etc/unidesk-gc
+          journalTargetBytes: 256MiB
+          tmpMinAgeHours: 24
+          includeJournal: false
+          includeTmp: false
+          includeAptCache: false
+          includeToolCaches: false
+          includeWebObserveArtifacts: false
+          includeK3sImageCache: false
+          includeHostContainerdCache: false
+          includeLocalPathOrphans: false
+          includeKubernetesObjectRetention: true
       JD01:
         hostDockerGc:
           dockerJsonLogs:
@@ -249,12 +328,15 @@ gc:
           configDir: /etc/unidesk-gc
           journalTargetBytes: 256MiB
           tmpMinAgeHours: 24
+          includeJournal: true
+          includeTmp: true
           includeAptCache: true
           includeToolCaches: false
           includeWebObserveArtifacts: true
           includeK3sImageCache: true
           includeHostContainerdCache: true
           includeLocalPathOrphans: true
+          includeKubernetesObjectRetention: false
           agentrunSessionPvcs:
             enabled: true
             namespace: agentrun-v02

config/unidesk-host-k8s.yaml

@@ -162,6 +162,7 @@ services:
       secretKey: DATABASE_URL
     storage:
       primary: github-repo
+      healthRefreshSeconds: 300
       github:
         repo: pikasTech/decision-center-data
         sshUrl: git@github.com:pikasTech/decision-center-data.git

docs/MDTODO/details/web-observe-runtime-reliability/R1.2.1_Task_Report.md

@@ -0,0 +1,28 @@
+# R1.2.1 任务报告
+
+## 结论
+
+已为 NC01 `devops-infra` 增加 YAML-first Kubernetes 历史对象留存。手动 `retention plan|run|status` 与 systemd policy timer 复用 `scripts/src/gc-remote-kubernetes-retention.py` 的同一个 fail-closed 筛选器；只删除终态、超过保留窗口、分组可识别且不属于每组最新保留数量的 PipelineRun，通过 Background ownerRef 级联回收下游对象。
+
+## 配置与保护
+
+- 配置真相：`config/unidesk-cli.yaml#gc.remote.targets.NC01.kubernetesObjectRetention` 与 `.policyTimer`。
+- 手动单批上限 500，自动单批上限 100，自动周期每 15 分钟。
+- 固定保护活跃、排队、过新、未知分组、时间不可解析、扫描截断和 Table 形状漂移对象。
+- 不直接删除 TaskRun、Pod、Secret、Event、PVC/PV 或 runtime workload。
+- NC01 policy 仅启用 `kubernetes-object-retention`，journal、tmp、cache、image 和 PVC 阶段全部关闭。
+
+## 真实运行证据
+
+- 初始 `devops-infra`：PipelineRun 1923、TaskRun 1922、Pod 932、Secret 1613。
+- 三个异步 job 分别成功删除 500、500、484 个 PipelineRun，全部批次成功。
+- 终态 overview：PipelineRun 451、TaskRun 450、Pod 444、Secret 455；剩余对象主要为活跃、24 小时内对象和每组最新保留对象。
+- timer `unidesk-devops-infra-retention.timer` 已 `loaded/active/waiting`，配置、runner 和状态文件均由 YAML 渲染。
+- 首次自动轮次 `failedCount=0`，只由 retention 阶段回收 6 个到龄 PipelineRun，下一轮已自动排期。
+- Python `py_compile`、`git diff --check` 与 `bun scripts/cli.ts check --syntax-only` 通过；未运行 Vitest。
+
+## 内存判定
+
+- `memory-distribution` 只以 `/proc/meminfo` 的 `MemAvailable` 判定，swap 单独披露。
+- 清理后 `MemAvailable` 约 2.8 GiB，距离 4 GiB 仍差约 1.25 GiB；WebProbe 不具备启动资格，未启动 Chromium。
+- 剩余压力主要来自 k3s 常驻 PSS、Kafka/Tempo/Tekton 控制面和 Decision Center zombie；不得继续扩大为 raw delete、k3s 重启或无证据 reclaim。

docs/MDTODO/web-observe-runtime-reliability.md

@@ -16,6 +16,9 @@
 
 实现长期修复：https://github.com/pikasTech/unidesk/issues/1842；补齐所有调用方终态 stop，runner 从 owning YAML 获取最大运行时长/max samples/artifact cap 并自停，GC snapshot/plan/status 披露 active/stale、Chrome RSS、last cleanup 与 stop failure，完成任务后将详细报告写入[任务报告](./details/web-observe-runtime-reliability/R1.2_Task_Report.md)。
 
+#### R1.2.1 [completed]
+
+为 UniDesk #1710 增加 `devops-infra` Kubernetes 历史对象留存：手动 `gc remote <node> retention plan|run|status` 与 YAML 定时滚动回收必须复用同一 fail-closed 筛选器；只处理终态 PipelineRun，通过 ownerRef 后台级联，保留活跃、过新、未知分组和每组最新对象；完成 dry-run、受控执行、timer 安装和 `MemAvailable` 实证后收口，完成任务后将详细报告写入[任务报告](./details/web-observe-runtime-reliability/R1.2.1_Task_Report.md)。
 ### R1.3
 
 测试与验收：https://github.com/pikasTech/unidesk/issues/1842；覆盖生命周期、stale/protected 分类和失败路径，提交独立 PR；合并后只等待自动 CI/CD 发布链，使用正式入口验证 observer/Chrome 退出、内存压力缓解且健康运行面不受扰动，完成任务后将详细报告写入[任务报告](./details/web-observe-runtime-reliability/R1.3_Task_Report.md)。

docs/reference/gc.md

@@ -30,6 +30,14 @@
   - 只读读取 `/proc/meminfo` 的 `MemAvailable`；
   - 不执行 cluster preflight、候选扫描或 mutation；
   - 固定披露 `metric=MemAvailable`、`source=/proc/meminfo`、`swapExcluded=true` 和可用字节数。
+- `gc remote <providerId> memory-distribution`：
+  - 只读输出物理内存、PSS、cgroup、zombie 和 Kubernetes 对象分布；
+  - 默认返回紧凑 overview，`--full` 才展开完整明细；
+  - swap 单独披露，不计入 WebProbe 启动资格。
+- `gc remote <providerId> retention plan|run --confirm|status --job-id <id>`：
+  - 按目标节点 `kubernetesObjectRetention` YAML 规划和异步回收 Kubernetes 历史对象；
+  - `plan` 与手动、自动 `run` 必须复用同一个 fail-closed 筛选器；
+  - 表格形状、时间或分组无法识别时保护对象并停止删除。
 - `--include-tool-caches`：本机和远端都必须是显式 opt-in，只清理固定 allowlist 的可重建 npm/npx/Bun 缓存；不得扩大成清理 `~/.npm`、`~/.bun`、`node_modules`、auth/config 或运行面依赖。
 
 所有成功和失败输出都必须是 JSON。`plan` 必须标记 `dryRun=true`、`mutation=false`；`run` 必须要求 `--confirm` 并报告 `diskBefore`、`diskAfter`、`summary`、`results` 和 `protected`。远端 GC 可用 `--target-use-percent N` 显式表达目标根盘水位；`summary.target` 必须给出目标所需释放量、候选估算、预计水位、缺口和 `safeStop` 决策，避免靠人工心算判断是否应该继续扩大清理范围。
@@ -104,6 +112,25 @@ PK01 是腾讯云 Docker provider，不是 G14 k3s/registry 节点；长期运
 
 PK01 pikanode temp retention 只允许清理 `/home/ubuntu/pikanode/html/temp` 下超过保留窗口的直接子目录，并必须保护 `html/download/`、`html/upload/`、`files/`、证书、Git state、直接日志文件和近期 temp workspace。该策略已固化为 PK01 节点本地 systemd timer 与 logrotate；人工排障时优先查看 `systemctl status unidesk-pk01-pikanode-temp-gc.timer` 和 `/var/log/unidesk-pk01/pikanode-temp-gc.log`。如果 PK01 高水位仍无法通过 temp retention 和通用低风险 GC 降下来，必须停止并进入 pikanode 下载产物留存、Docker image retention 或容量扩容决策，不能把 `download/`、`files/` 或 Docker overlay 当作普通临时目录删除。
 
+## NC01 远端策略
+
+- NC01 的 `devops-infra` Tekton 历史对象留存由以下 YAML 唯一拥有：
+  - 手动筛选策略位于 `config/unidesk-cli.yaml#gc.remote.targets.NC01.kubernetesObjectRetention`；
+  - 自动周期、单批上限和启用阶段位于 `config/unidesk-cli.yaml#gc.remote.targets.NC01.policyTimer`。
+- 允许回收的对象必须同时满足：
+  - 资源类型是 YAML 指定的 PipelineRun；
+  - `Succeeded` 为 `True` 或 `False`；
+  - 对象年龄不小于 YAML 保留窗口；
+  - 分组标签完整，并且不属于每组最新保留数量。
+- 以下情况固定保护：
+  - 活跃、排队或状态未知；
+  - 过新、时间不可解析或分组未知；
+  - API 扫描截断、Table 形状漂移或命令超时。
+- 执行只删除 PipelineRun，并使用 YAML 指定的 ownerRef 级联策略：
+  - 不直接删除 TaskRun、Pod、Secret 或 Event；
+  - 手动入口提交异步 job 后必须用 `status --job-id` 查询终态；
+  - 自动 systemd timer 复用同一筛选器，并把每轮结果写入 YAML 指定的状态文件。
+
 ## JD01 远端策略
 
 JD01 是 YAML-first k3s provider，承载 AgentRun、HWLAB v0.3、Web probe sentinel 和相关 PVC/state artifact。`gc remote JD01 ...` 的目标节点、lane、namespace 集合、state root、保护路径、扫描预算、artifact cap、retention 窗口、observer stop 策略和输出 limit 必须来自 `config/unidesk-cli.yaml#gc.remote.targets.JD01` 或等价 source of truth；CLI 参数只作为一次性覆盖，不得把 JD01 namespace、路径或阈值写成脚本隐藏默认。

scripts/native/deploy/render-unidesk-host-k8s.mjs

@@ -206,15 +206,22 @@ function readServiceStorage(service, servicePath) {
   if (storage === null) return null;
   const primary = required(storage.primary, `${servicePath}.storage.primary`);
   if (primary !== "postgres" && primary !== "github-repo") throw new Error(`${servicePath}.storage.primary must be postgres or github-repo`);
+  const healthRefreshSeconds = storage.healthRefreshSeconds === undefined
+    ? null
+    : positiveInteger(storage.healthRefreshSeconds, `${servicePath}.storage.healthRefreshSeconds`);
+  const resolved = {
+    primary,
+    ...(healthRefreshSeconds === null ? {} : { healthRefreshSeconds }),
+  };
   const github = primary === "github-repo" ? record(storage.github, `${servicePath}.storage.github`) : null;
-  if (github === null) return { primary };
+  if (github === null) return resolved;
   const ssh = record(github.ssh, `${servicePath}.storage.github.ssh`);
   const privateKey = record(ssh.privateKey, `${servicePath}.storage.github.ssh.privateKey`);
   const knownHosts = record(ssh.knownHosts, `${servicePath}.storage.github.ssh.knownHosts`);
   const privateKeySource = readFileSource(record(privateKey.sourceRef, `${servicePath}.storage.github.ssh.privateKey.sourceRef`), `${servicePath}.storage.github.ssh.privateKey.sourceRef`);
   const knownHostsSource = readFileSource(record(knownHosts.sourceRef, `${servicePath}.storage.github.ssh.knownHosts.sourceRef`), `${servicePath}.storage.github.ssh.knownHosts.sourceRef`);
   return {
-    primary,
+    ...resolved,
     github: {
       repo: required(github.repo, `${servicePath}.storage.github.repo`),
       sshUrl: required(github.sshUrl, `${servicePath}.storage.github.sshUrl`),
@@ -629,6 +636,12 @@ function managedStorageEnv(envPrefix, storage) {
     `- name: ${envPrefix}_STORAGE_PRIMARY`,
     `  value: ${yamlScalar(storage?.primary ?? "postgres")}`,
   ];
+  if (storage?.healthRefreshSeconds !== undefined) {
+    lines.push(
+      `- name: ${envPrefix}_STORAGE_HEALTH_REFRESH_MS`,
+      `  value: ${yamlScalar(positiveInteger(storage.healthRefreshSeconds, "storage.healthRefreshSeconds") * 1000)}`,
+    );
+  }
   if (storage?.github === undefined) return lines.join("\n");
   lines.push(
     `- name: ${envPrefix}_GIT_REPO`,

scripts/src/gc-remote-kubernetes-retention.py

@@ -0,0 +1,490 @@
+import hashlib
+import json
+import re
+import subprocess
+import time
+import urllib.parse
+
+
+def kubernetes_retention_config(config):
+    if not isinstance(config, dict):
+        raise RuntimeError("required YAML kubernetesObjectRetention policy is missing")
+    if not isinstance(config.get("enabled"), bool):
+        raise RuntimeError("kubernetesObjectRetention.enabled must be a YAML boolean")
+    required_strings = [
+        "configId",
+        "kubeconfigPath",
+        "namespace",
+        "resource",
+        "listMode",
+        "countPath",
+        "terminalConditionType",
+        "propagationPolicy",
+    ]
+    for key in required_strings:
+        value = config.get(key)
+        if not isinstance(value, str) or not value:
+            raise RuntimeError("kubernetesObjectRetention.%s must be a non-empty YAML string" % key)
+    if not config.get("kubeconfigPath").startswith("/"):
+        raise RuntimeError("kubernetesObjectRetention.kubeconfigPath must be absolute")
+    if not re.match(r"^[a-z0-9]([-a-z0-9]*[a-z0-9])?$", config.get("namespace")):
+        raise RuntimeError("kubernetesObjectRetention.namespace is invalid")
+    if config.get("listMode") != "server-table":
+        raise RuntimeError("kubernetesObjectRetention.listMode must be server-table")
+    if "{namespace}" not in config.get("countPath") or not config.get("countPath").startswith("/apis/"):
+        raise RuntimeError("kubernetesObjectRetention.countPath must be a namespaced API path template")
+    if config.get("propagationPolicy") not in ["Background", "Foreground", "Orphan"]:
+        raise RuntimeError("kubernetesObjectRetention.propagationPolicy is invalid")
+    positive_integers = [
+        "minAgeHours",
+        "keepLatestPerGroup",
+        "manualMaxDeletePerRun",
+        "automaticMaxDeletePerRun",
+        "previewLimit",
+        "scanLimit",
+        "requestTimeoutSeconds",
+        "deleteBatchSize",
+        "deleteTimeoutSeconds",
+    ]
+    for key in positive_integers:
+        value = config.get(key)
+        if isinstance(value, bool) or not isinstance(value, int) or value <= 0:
+            raise RuntimeError("kubernetesObjectRetention.%s must be a positive YAML integer" % key)
+    settle_seconds = config.get("postDeleteSettleSeconds")
+    if isinstance(settle_seconds, bool) or not isinstance(settle_seconds, int) or settle_seconds < 0:
+        raise RuntimeError("kubernetesObjectRetention.postDeleteSettleSeconds must be a non-negative YAML integer")
+    terminal_statuses = config.get("terminalStatuses")
+    if not isinstance(terminal_statuses, list) or not terminal_statuses or any(
+        not isinstance(item, str) or not item for item in terminal_statuses
+    ):
+        raise RuntimeError("kubernetesObjectRetention.terminalStatuses must be a non-empty YAML string list")
+    selectors = config.get("groupSelectors")
+    if not isinstance(selectors, list) or not selectors:
+        raise RuntimeError("kubernetesObjectRetention.groupSelectors must be a non-empty YAML list")
+    selector_ids = set()
+    for selector in selectors:
+        if not isinstance(selector, dict):
+            raise RuntimeError("kubernetesObjectRetention.groupSelectors items must be YAML objects")
+        selector_id = selector.get("id")
+        labels = selector.get("groupByLabels")
+        if not isinstance(selector_id, str) or not re.match(r"^[a-z0-9][a-z0-9-]{0,63}$", selector_id):
+            raise RuntimeError("kubernetesObjectRetention group selector id is invalid")
+        if selector_id in selector_ids:
+            raise RuntimeError("kubernetesObjectRetention group selector ids must be unique")
+        selector_ids.add(selector_id)
+        if not isinstance(labels, list) or not labels or any(
+            not isinstance(item, str) or not item or len(item) > 253 for item in labels
+        ):
+            raise RuntimeError("kubernetesObjectRetention groupByLabels must be a non-empty YAML string list")
+    return config
+
+
+def kubernetes_retention_command(config, args, timeout_key):
+    started = time.time()
+    timeout = int(config.get(timeout_key))
+    try:
+        process = subprocess.run(
+            args,
+            text=True,
+            stdout=subprocess.PIPE,
+            stderr=subprocess.PIPE,
+            timeout=timeout,
+        )
+        return {
+            "exitCode": process.returncode,
+            "stdout": process.stdout,
+            "stderr": process.stderr,
+            "timedOut": False,
+            "elapsedMs": int((time.time() - started) * 1000),
+        }
+    except subprocess.TimeoutExpired as exc:
+        stdout = exc.stdout.decode("utf-8", errors="replace") if isinstance(exc.stdout, bytes) else (exc.stdout or "")
+        stderr = exc.stderr.decode("utf-8", errors="replace") if isinstance(exc.stderr, bytes) else (exc.stderr or "")
+        return {
+            "exitCode": 124,
+            "stdout": stdout,
+            "stderr": stderr,
+            "timedOut": True,
+            "elapsedMs": int((time.time() - started) * 1000),
+        }
+
+
+def kubernetes_retention_command_summary(result):
+    return {
+        "exitCode": result.get("exitCode"),
+        "timedOut": result.get("timedOut") is True,
+        "elapsedMs": result.get("elapsedMs"),
+        "stdoutTail": str(result.get("stdout") or "")[-800:],
+        "stderrTail": str(result.get("stderr") or "")[-800:],
+    }
+
+
+def kubernetes_retention_duration_hours(value):
+    if not isinstance(value, str) or not value or value in ["<none>", "<unknown>"]:
+        return None
+    matches = list(re.finditer(r"([0-9]+)([ysmhd])", value))
+    if not matches or "".join(match.group(0) for match in matches) != value:
+        return None
+    multipliers = {"y": 8760.0, "s": 1.0 / 3600.0, "m": 1.0 / 60.0, "h": 1.0, "d": 24.0}
+    return sum(int(match.group(1)) * multipliers[match.group(2)] for match in matches)
+
+
+def kubernetes_retention_group(labels, selectors):
+    for selector in selectors:
+        keys = selector.get("groupByLabels") or []
+        values = [labels.get(key) for key in keys]
+        if all(isinstance(value, str) and value for value in values):
+            return {
+                "selectorId": selector.get("id"),
+                "key": "%s:%s" % (selector.get("id"), "|".join(values)),
+                "labels": {key: labels.get(key) for key in keys},
+            }
+    return None
+
+
+def kubernetes_retention_row(item, config, now_timestamp):
+    labels = item.get("labels") or {}
+    terminal_status = str(item.get("terminalStatus") or "")
+    terminal = terminal_status in set(config.get("terminalStatuses") or [])
+    group = kubernetes_retention_group(labels, config.get("groupSelectors") or [])
+    age_hours = kubernetes_retention_duration_hours(item.get("age"))
+    return {
+        "name": item.get("name"),
+        "createdAt": None,
+        "createdTimestamp": now_timestamp - age_hours * 3600.0 if age_hours is not None else None,
+        "age": item.get("age"),
+        "ageHours": round(age_hours, 2) if age_hours is not None else None,
+        "terminal": terminal,
+        "terminalStatus": terminal_status or None,
+        "terminalReason": item.get("terminalReason"),
+        "group": group,
+    }
+
+
+def kubernetes_retention_list(config):
+    scan_limit = int(config.get("scanLimit"))
+    result = kubernetes_retention_command(config, [
+        "kubectl",
+        "--kubeconfig", config.get("kubeconfigPath"),
+        "--namespace", config.get("namespace"),
+        "get", config.get("resource"),
+        "--no-headers=true",
+        "--show-labels=true",
+    ], "requestTimeoutSeconds")
+    command_summary = kubernetes_retention_command_summary(result)
+    if result.get("exitCode") != 0:
+        return {
+            "ok": False,
+            "error": "kubernetes-retention-list-failed",
+            "items": [],
+            "scan": {"scannedCount": 0, "limit": scan_limit, "truncated": True, "commands": [command_summary]},
+        }
+    items = []
+    invalid_rows = []
+    sparse_non_terminal_rows = 0
+    terminal_statuses = set(config.get("terminalStatuses") or [])
+    for line_number, line in enumerate((result.get("stdout") or "").splitlines(), start=1):
+        if not line.strip():
+            continue
+        parts = line.split()
+        terminal = len(parts) >= 2 and parts[1] in terminal_statuses
+        valid_column_count = len(parts) == 6 if terminal else 4 <= len(parts) <= 6
+        if not valid_column_count:
+            invalid_rows.append({
+                "line": line_number,
+                "columnCount": len(parts),
+                "rowPreview": line[:600],
+            })
+            continue
+        if not terminal and len(parts) < 6:
+            sparse_non_terminal_rows += 1
+        labels = {}
+        labels_field = parts[-1]
+        if labels_field not in ["<none>", ""]:
+            for label in labels_field.split(","):
+                key, separator, value = label.partition("=")
+                if separator and key and value:
+                    labels[key] = value
+            if not labels:
+                invalid_rows.append({
+                    "line": line_number,
+                    "columnCount": len(parts),
+                    "rowPreview": line[:600],
+                    "reason": "labels-column-invalid",
+                })
+                continue
+        age = parts[3] if len(parts) >= 5 else None
+        completion_age = parts[4] if len(parts) == 6 else None
+        items.append({
+            "name": parts[0],
+            "terminalStatus": parts[1],
+            "terminalReason": parts[2],
+            "age": age,
+            "completionAge": completion_age,
+            "labels": labels,
+        })
+    truncated = len(items) > scan_limit
+    if truncated:
+        items = items[:scan_limit]
+    if invalid_rows:
+        return {
+            "ok": False,
+            "error": "kubernetes-retention-table-shape-invalid",
+            "items": [],
+            "scan": {
+                "scannedCount": len(items),
+                "limit": scan_limit,
+                "truncated": True,
+                "invalidRows": invalid_rows[:10],
+                "sparseNonTerminalRowCount": sparse_non_terminal_rows,
+                "commands": [command_summary],
+            },
+        }
+    return {
+        "ok": not truncated,
+        "error": "kubernetes-retention-scan-limit-reached" if truncated else None,
+        "items": items,
+        "scan": {
+            "scannedCount": len(items),
+            "totalHint": len(items),
+            "limit": scan_limit,
+            "mode": config.get("listMode"),
+            "truncated": truncated,
+            "sparseNonTerminalRowCount": sparse_non_terminal_rows,
+            "commands": [command_summary],
+        },
+    }
+
+
+def kubernetes_retention_plan(config, mode, limit_override=None, full=False):
+    config = kubernetes_retention_config(config)
+    if not config.get("enabled"):
+        return {
+            "ok": False,
+            "error": "kubernetes-object-retention-disabled",
+            "configId": config.get("configId"),
+            "mutation": False,
+        }
+    listed = kubernetes_retention_list(config)
+    if not listed.get("ok"):
+        return {
+            "ok": False,
+            "error": listed.get("error"),
+            "message": listed.get("message"),
+            "configId": config.get("configId"),
+            "namespace": config.get("namespace"),
+            "mutation": False,
+            "scan": listed.get("scan"),
+        }
+    now_timestamp = time.time()
+    rows = [kubernetes_retention_row(item, config, now_timestamp) for item in listed.get("items") or []]
+    rows = [item for item in rows if isinstance(item.get("name"), str) and item.get("name")]
+    groups = {}
+    for row in rows:
+        group = row.get("group")
+        if group is not None:
+            groups.setdefault(group.get("key"), []).append(row)
+    protected_latest = set()
+    for group_rows in groups.values():
+        terminal_rows = [item for item in group_rows if item.get("terminal")]
+        terminal_rows.sort(key=lambda item: item.get("createdTimestamp") or 0, reverse=True)
+        for item in terminal_rows[:int(config.get("keepLatestPerGroup"))]:
+            protected_latest.add(item.get("name"))
+    eligible = []
+    protected = {
+        "active": 0,
+        "young": 0,
+        "latestPerGroup": 0,
+        "unknownGroup": 0,
+        "invalidTimestamp": 0,
+    }
+    for row in rows:
+        if not row.get("terminal"):
+            protected["active"] += 1
+        elif row.get("group") is None:
+            protected["unknownGroup"] += 1
+        elif row.get("createdTimestamp") is None or row.get("ageHours") is None:
+            protected["invalidTimestamp"] += 1
+        elif row.get("ageHours") < int(config.get("minAgeHours")):
+            protected["young"] += 1
+        elif row.get("name") in protected_latest:
+            protected["latestPerGroup"] += 1
+        else:
+            eligible.append(row)
+    eligible.sort(key=lambda item: item.get("createdTimestamp") or 0)
+    configured_limit = int(config.get("automaticMaxDeletePerRun" if mode == "automatic" else "manualMaxDeletePerRun"))
+    effective_limit = configured_limit
+    if limit_override is not None:
+        effective_limit = min(configured_limit, max(1, int(limit_override)))
+    selected = eligible[:effective_limit]
+    selected_names = [item.get("name") for item in selected]
+    fingerprint = hashlib.sha256("\n".join(selected_names).encode("utf-8")).hexdigest()
+    group_summary = []
+    for group_key, group_rows in groups.items():
+        group_selected = [item for item in selected if (item.get("group") or {}).get("key") == group_key]
+        group_summary.append({
+            "groupKey": group_key,
+            "totalCount": len(group_rows),
+            "terminalCount": len([item for item in group_rows if item.get("terminal")]),
+            "selectedCount": len(group_selected),
+        })
+    group_summary.sort(key=lambda item: item.get("totalCount"), reverse=True)
+    preview_limit = int(config.get("previewLimit"))
+    preview = selected if full else selected[:preview_limit]
+    candidates = [
+        {
+            "name": item.get("name"),
+            "createdAt": item.get("createdAt"),
+            "age": item.get("age"),
+            "ageHours": item.get("ageHours"),
+            "terminalStatus": item.get("terminalStatus"),
+            "terminalReason": item.get("terminalReason"),
+            "group": item.get("group"),
+        }
+        for item in preview
+    ]
+    return {
+        "ok": True,
+        "configId": config.get("configId"),
+        "mode": mode,
+        "namespace": config.get("namespace"),
+        "resource": config.get("resource"),
+        "dryRun": True,
+        "mutation": False,
+        "criteria": {
+            "terminalConditionType": config.get("terminalConditionType"),
+            "terminalStatuses": config.get("terminalStatuses"),
+            "minAgeHours": config.get("minAgeHours"),
+            "keepLatestPerGroup": config.get("keepLatestPerGroup"),
+            "effectiveDeleteLimit": effective_limit,
+            "propagationPolicy": config.get("propagationPolicy"),
+        },
+        "scan": listed.get("scan"),
+        "summary": {
+            "objectCount": len(rows),
+            "terminalCount": len([item for item in rows if item.get("terminal")]),
+            "eligibleCount": len(eligible),
+            "selectedCount": len(selected),
+            "omittedEligibleCount": max(0, len(eligible) - len(selected)),
+            "protected": protected,
+            "groupCount": len(groups),
+        },
+        "groups": group_summary if full else group_summary[:preview_limit],
+        "candidates": candidates,
+        "returnedCandidateCount": len(candidates),
+        "omittedCandidateCount": max(0, len(selected) - len(candidates)),
+        "selectedNames": selected_names,
+        "planFingerprint": fingerprint,
+    }
+
+
+def kubernetes_retention_count(config):
+    base_path = config.get("countPath").replace(
+        "{namespace}", urllib.parse.quote(config.get("namespace"), safe="")
+    )
+    separator = "&" if "?" in base_path else "?"
+    result = kubernetes_retention_command(config, [
+        "kubectl", "--kubeconfig", config.get("kubeconfigPath"), "get", "--raw", "%s%slimit=1" % (base_path, separator),
+    ], "requestTimeoutSeconds")
+    command_summary = kubernetes_retention_command_summary(result)
+    command_summary.pop("stdoutTail", None)
+    if result.get("exitCode") != 0:
+        return {"ok": False, "command": command_summary}
+    try:
+        payload = json.loads(result.get("stdout") or "{}")
+        metadata = payload.get("metadata") or {}
+        return {
+            "ok": True,
+            "count": len(payload.get("items") or []) + int(metadata.get("remainingItemCount") or 0),
+            "command": command_summary,
+        }
+    except (TypeError, ValueError, json.JSONDecodeError) as exc:
+        return {"ok": False, "error": str(exc), "command": command_summary}
+
+
+def kubernetes_retention_execute(config, mode, limit_override=None, progress=None, full=False):
+    config = kubernetes_retention_config(config)
+    before = kubernetes_retention_count(config)
+    plan = kubernetes_retention_plan(config, mode, limit_override=limit_override, full=True)
+    if not plan.get("ok"):
+        return {"ok": False, "plan": plan, "before": before, "mutation": False, "deletedCount": 0}
+    selected_names = list(plan.get("selectedNames") or [])
+    if not selected_names:
+        return {"ok": True, "plan": plan, "before": before, "after": before, "mutation": False, "deletedCount": 0, "batches": []}
+    batch_size = int(config.get("deleteBatchSize"))
+    batches = []
+    deleted_count = 0
+    for offset in range(0, len(selected_names), batch_size):
+        names = selected_names[offset:offset + batch_size]
+        result = kubernetes_retention_command(config, [
+            "kubectl",
+            "--kubeconfig", config.get("kubeconfigPath"),
+            "--namespace", config.get("namespace"),
+            "delete", config.get("resource"),
+            "--ignore-not-found=true",
+            "--wait=false",
+            "--cascade=%s" % config.get("propagationPolicy").lower(),
+        ] + names, "deleteTimeoutSeconds")
+        ok = result.get("exitCode") == 0
+        if ok:
+            deleted_count += len(names)
+        batches.append({
+            "index": len(batches),
+            "requestedCount": len(names),
+            "ok": ok,
+            "command": kubernetes_retention_command_summary(result),
+        })
+        if progress is not None:
+            progress(deleted_count, len(selected_names))
+        if not ok:
+            break
+    settle_seconds = int(config.get("postDeleteSettleSeconds"))
+    if settle_seconds > 0:
+        time.sleep(settle_seconds)
+    after = kubernetes_retention_count(config)
+    ok = all(item.get("ok") for item in batches)
+    compact_plan = dict(plan)
+    if not full:
+        compact_plan.pop("selectedNames", None)
+        compact_plan["candidates"] = compact_plan.get("candidates", [])[:int(config.get("previewLimit"))]
+    return {
+        "ok": ok,
+        "configId": config.get("configId"),
+        "mode": mode,
+        "namespace": config.get("namespace"),
+        "resource": config.get("resource"),
+        "mutation": deleted_count > 0,
+        "requestedDeleteCount": len(selected_names),
+        "deletedCount": deleted_count,
+        "before": before,
+        "after": after,
+        "objectCountDelta": (
+            int(before.get("count")) - int(after.get("count"))
+            if before.get("ok") and after.get("ok") else None
+        ),
+        "plan": compact_plan,
+        "batches": batches,
+    }
+
+
+def run_kubernetes_object_retention_policy(config):
+    retention = config.get("kubernetesObjectRetention") if isinstance(config.get("kubernetesObjectRetention"), dict) else {}
+    if not retention.get("enabled"):
+        return {"id": "kubernetes-object-retention", "ok": True, "skipped": True}
+    result = kubernetes_retention_execute(retention, "automatic", full=False)
+    return {
+        "id": "kubernetes-object-retention",
+        "ok": result.get("ok") is True,
+        "skipped": result.get("mutation") is not True,
+        "configId": result.get("configId"),
+        "namespace": result.get("namespace"),
+        "resource": result.get("resource"),
+        "candidateCount": ((result.get("plan") or {}).get("summary") or {}).get("eligibleCount"),
+        "requestedDeleteCount": result.get("requestedDeleteCount"),
+        "deletedCount": result.get("deletedCount"),
+        "objectCountDelta": result.get("objectCountDelta"),
+        "before": result.get("before"),
+        "after": result.get("after"),
+        "batches": result.get("batches"),
+        "error": ((result.get("plan") or {}).get("error")) if not result.get("ok") else None,
+    }

scripts/src/gc-remote-memory-distribution.py

@@ -0,0 +1,474 @@
+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),
+        "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 collect_memory_distribution_processes(config):
+    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, "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")]
+    pss_limit = memory_distribution_limit(config, "processPssLimit")
+    for index, item in enumerate(top):
+        item["cgroupPath"] = memory_distribution_cgroup_for_pid(item.get("pid"))
+        if index < pss_limit:
+            item.update(memory_distribution_pss(item.get("pid")) or {})
+    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 {}
+        parent_rows.append({
+            "ppid": ppid,
+            "parentComm": parent.get("comm"),
+            "zombieCount": count,
+            "cgroupPath": memory_distribution_cgroup_for_pid(ppid),
+        })
+    return {
+        "ok": True,
+        "command": memory_distribution_command_summary(result),
+        "processCount": len(rows),
+        "rssTotalBytes": sum(safe_int(item.get("rssBytes")) for item in rows),
+        "top": top,
+        "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 ["path", "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
+        }
+    return compact
+
+def compact_memory_distribution_process(item):
+    return {
+        key: item.get(key)
+        for key in ["pid", "comm", "rssBytes", "pssBytes", "pssAnonBytes", "swapPssBytes", "cgroupPath"]
+        if item.get(key) is not None
+    }
+
+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 []:
+        by_parent.append({
+            key: item.get(key)
+            for key in ["ppid", "parentComm", "zombieCount"]
+            if item.get(key) is not None
+        })
+    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)
+    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"] = {
+        "snapshot": host_snapshot,
+        "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"),
+        "top": [compact_memory_distribution_process(item) for item in processes.get("top") or []],
+        "zombies": compact_memory_distribution_zombies(processes.get("zombies") or {}),
+    }
+    payload["kubernetes"] = {
+        "objects": metrics.get("objects"),
+        "namespaceCounts": namespace_counts,
+        "command": metrics.get("command"),
+    }
+    payload["collection"] = {
+        "configSource": payload.get("collection", {}).get("configSource"),
+        "podRuntimeMapCount": len(pod_names),
+        "dockerRuntimeMapCount": len(docker_names),
+        "runtimeMaps": runtime_commands,
+        "full": False,
+    }
+    return payload

scripts/src/gc-remote-memory-pressure.test.ts

@@ -121,3 +121,41 @@ test("memory-pressure async runner uses a parent-child barrier and locked worker
   assert.match(runnerSource, /workerAlive/u);
   assert.match(runnerSource, /worker-identity-unavailable|worker-exited-without-terminal-state/u);
 });
+
+test("memory-pressure plan includes only YAML allowlisted cgroup reclaim stages with swap reserve", () => {
+  const result = runPythonFixture(String.raw`
+MEMORY_CONFIG = {
+  "observeStateRoots": ["/tmp/observe"], "minimumReadableObserveRoots": 1,
+  "observeRunDepth": 6, "observeScanLimit": 100, "staleRunMaxAgeHours": 1,
+  "processScanLimit": 100, "processStaleMinAgeSeconds": 3600,
+  "maxMemoryReclaimCandidates": 20, "terminationGraceSeconds": 1,
+  "terminationPollMilliseconds": 50, "observerRootPatterns": ["observer-runner.mjs"],
+  "browserRootPatterns": ["cliDaemon.mjs"], "browserProcessPatterns": ["chromium"],
+  "cgroupReclaim": {
+    "enabled": True, "targetMemAvailableBytes": 4294967296,
+    "minimumSwapFreeBytesAfterReclaim": 402653184,
+    "targets": [
+      {"id": "file-pages", "path": "/sys/fs/cgroup/system.slice/k3s.service", "reclaimBytes": 402653184, "swappiness": 0, "priority": 10},
+      {"id": "inactive-anon", "path": "/sys/fs/cgroup/system.slice/k3s.service", "reclaimBytes": 1342177280, "swappiness": 100, "priority": 20},
+    ],
+  },
+}
+read_proc_process_table = lambda: {1: {"pid": 1, "ppid": 0, "sid": 1, "startTicks": 1, "ageSeconds": 1, "rssBytes": 1, "commandLine": "init", "commandPreview": "init"}}
+observer_run_records = lambda: []
+collect_memory_snapshot = lambda: {"ok": True, "memory": {"availableBytes": 3 * 1024 * 1024 * 1024}, "swap": {"freeBytes": 2 * 1024 * 1024 * 1024}}
+os.path.realpath = lambda path: path
+os.path.isdir = lambda path: True
+os.path.islink = lambda path: False
+os.path.exists = lambda path: True
+os.access = lambda path, mode: True
+read_cgroup_integer = lambda path, name: 6 * 1024 * 1024 * 1024
+read_cgroup_memory_stat = lambda path: {"inactive_anon": 2 * 1024 * 1024 * 1024}
+candidates = collect_memory_reclaim_candidates()
+print(json.dumps({"ids": [item["id"] for item in candidates], "swappiness": [item["swappiness"] for item in candidates], "diagnostics": memory_reclaim_scan_diagnostics()}))
+`);
+  assert.deepEqual(result.ids, ["cgroup-memory-reclaim:file-pages", "cgroup-memory-reclaim:inactive-anon"]);
+  assert.deepEqual(result.swappiness, [0, 100]);
+  assert.equal(((result.diagnostics as Record<string, unknown>).cgroupReclaim as Record<string, unknown>).eligibleCount, 2);
+  assert.match(webObserveSource, /minimum SwapFree reserve/u);
+  assert.match(webObserveSource, /memory\.reclaim/u);
+});

scripts/src/gc-remote-policy-runner.py

@@ -8,6 +8,8 @@ import sys
 import time
 from datetime import datetime, timezone
 
+# __UNIDESK_GC_REMOTE_POLICY_KUBERNETES_RETENTION_HELPERS__
+
 
 def now_iso():
     return datetime.now(timezone.utc).replace(microsecond=0).isoformat().replace("+00:00", "Z")
@@ -135,6 +137,8 @@ def write_state(config, payload):
 
 
 def run_journal(config):
+    if config.get("includeJournal") is not True:
+        return {"id": "journal", "ok": True, "skipped": True}
     target = safe_int(config.get("journalTargetBytes"), 268435456)
     result = command(["journalctl", "--vacuum-size=%d" % target], 30)
     return {"id": "journal", "ok": result["exitCode"] == 0, "command": result}
@@ -150,6 +154,8 @@ def run_apt(config):
 
 
 def run_tmp(config):
+    if config.get("includeTmp") is not True:
+        return {"id": "tmp-allowlist", "ok": True, "skipped": True}
     prefixes = list(config.get("tmpPrefixAllowlist") or [])
     protected = set(config.get("tmpExactProtect") or [])
     cutoff = time.time() - float(config.get("tmpMinAgeHours") or 24) * 3600.0
@@ -181,6 +187,8 @@ def run_tmp(config):
 
 
 def run_tool_caches(config):
+    if config.get("includeToolCaches") is not True:
+        return {"id": "tool-caches", "ok": True, "skipped": True}
     paths = list(config.get("toolCachePaths") or [])
     reclaimed = 0
     deleted = 0
@@ -453,7 +461,7 @@ def main():
     config_path = sys.argv[1] if len(sys.argv) > 1 else ""
     config = load_config(config_path)
     results = []
-    for fn in [run_journal, run_apt, run_tmp, run_tool_caches, run_web_observe, run_session_pvcs, run_k3s_images, run_host_containerd, run_local_path_orphans]:
+    for fn in [run_journal, run_apt, run_tmp, run_tool_caches, run_web_observe, run_session_pvcs, run_k3s_images, run_host_containerd, run_local_path_orphans, run_kubernetes_object_retention_policy]:
         try:
             results.append(fn(config))
         except Exception as exc:

scripts/src/gc-remote-runner.py

@@ -23,6 +23,7 @@ PVC_CONFIG = REMOTE_TARGET.get("pvcAttribution") if isinstance(REMOTE_TARGET.get
 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 {}
+KUBERNETES_RETENTION_CONFIG = REMOTE_TARGET.get("kubernetesObjectRetention") if isinstance(REMOTE_TARGET.get("kubernetesObjectRetention"), 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")
@@ -348,6 +349,67 @@ def emit_json(payload, persist_large=True):
     compact = compact_payload_for_stdout(payload, full_size, job_id or None, paths)
     print(json.dumps(compact, ensure_ascii=False, indent=2))
 
+def compact_kubernetes_retention_result(result):
+    if not isinstance(result, dict):
+        return result
+    plan = result.get("plan") if isinstance(result.get("plan"), dict) else {}
+    scan = plan.get("scan") if isinstance(plan.get("scan"), dict) else {}
+    batches = []
+    for item in result.get("batches") or []:
+        if not isinstance(item, dict):
+            continue
+        command_payload = item.get("command") if isinstance(item.get("command"), dict) else {}
+        batches.append({
+            "index": item.get("index"),
+            "requestedCount": item.get("requestedCount"),
+            "ok": item.get("ok") is True,
+            "command": {
+                key: command_payload.get(key)
+                for key in ["exitCode", "timedOut", "elapsedMs"]
+                if command_payload.get(key) is not None
+            },
+        })
+    compact = {
+        key: result.get(key)
+        for key in [
+            "ok",
+            "configId",
+            "mode",
+            "namespace",
+            "resource",
+            "mutation",
+            "requestedDeleteCount",
+            "deletedCount",
+            "objectCountDelta",
+        ]
+        if result.get(key) is not None
+    }
+    for count_key in ["before", "after"]:
+        count_payload = result.get(count_key) if isinstance(result.get(count_key), dict) else {}
+        command_payload = count_payload.get("command") if isinstance(count_payload.get("command"), dict) else {}
+        compact[count_key] = {
+            key: count_payload.get(key)
+            for key in ["ok", "count", "error"]
+            if count_payload.get(key) is not None
+        }
+        compact[count_key]["command"] = {
+            key: command_payload.get(key)
+            for key in ["exitCode", "timedOut", "elapsedMs"]
+            if command_payload.get(key) is not None
+        }
+    compact["plan"] = {
+        "criteria": plan.get("criteria"),
+        "summary": plan.get("summary"),
+        "planFingerprint": plan.get("planFingerprint"),
+        "scan": {
+            key: scan.get(key)
+            for key in ["scannedCount", "totalHint", "limit", "mode", "truncated", "sparseNonTerminalRowCount"]
+            if scan.get(key) is not None
+        },
+    }
+    compact["batches"] = batches
+    return compact
+
 def remote_gc_job_status():
     job_id = job_id_or_none()
     if not job_id:
@@ -366,7 +428,8 @@ def remote_gc_job_status():
         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_scopes = set(["memory-pressure-only", "kubernetes-object-retention"])
+        if payload.get("scope") in worker_scopes 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
@@ -374,10 +437,15 @@ def remote_gc_job_status():
             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:
+                outcome = (
+                    "memory-reclaim-worker-exited-without-terminal-state"
+                    if payload.get("scope") == "memory-pressure-only"
+                    else "kubernetes-retention-worker-exited-without-terminal-state"
+                )
                 payload.update({
                     "ok": False,
                     "status": "failed",
-                    "outcome": "memory-reclaim-worker-exited-without-terminal-state",
+                    "outcome": outcome,
                     "workerAlive": False,
                     "workerObservedState": actual_identity.get("state") if actual_identity else None,
                     "workerObservedStartTicks": actual_start_ticks,
@@ -398,6 +466,8 @@ def remote_gc_job_status():
     finally:
         if descriptor is not None:
             release_job_lock(descriptor)
+    if payload.get("scope") == "kubernetes-object-retention" and not bool(OPTIONS.get("full")):
+        payload["retention"] = compact_kubernetes_retention_result(payload.get("retention"))
     payload["logTail"] = read_file_tail(paths["log"])
     return payload
 
@@ -666,14 +736,27 @@ def collect_memory_snapshot():
         "totalHuman": fmt_bytes(total_bytes) if total_bytes is not None else None,
         "availableHuman": fmt_bytes(available_bytes),
     }
+    swap_total_bytes = fields.get("SwapTotal")
+    swap_free_bytes = fields.get("SwapFree")
+    swap = {
+        "totalBytes": swap_total_bytes,
+        "freeBytes": swap_free_bytes,
+        "totalHuman": fmt_bytes(swap_total_bytes) if swap_total_bytes is not None else None,
+        "freeHuman": fmt_bytes(swap_free_bytes) if swap_free_bytes is not None else None,
+    }
     return {
         "ok": True,
         "metric": "MemAvailable",
         "source": source,
         "swapExcluded": True,
         "memory": memory,
+        "swap": swap,
     }
 
+# __UNIDESK_GC_REMOTE_MEMORY_DISTRIBUTION_HELPERS__
+
+# __UNIDESK_GC_REMOTE_KUBERNETES_RETENTION_HELPERS__
+
 def observe_run_record(path, stale_hours):
     stat = os.stat(path)
     heartbeat = read_json_file(os.path.join(path, "heartbeat.json")) or {}
@@ -1524,6 +1607,8 @@ def execute(candidate):
         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 == "cgroup-memory-reclaim":
+        return execute_cgroup_reclaim_candidate(candidate)
     if kind == "k3s-cri-image-prune":
         return execute_k3s_image_cache_prune()
     if kind == "host-containerd-cache-prune":
@@ -1577,6 +1662,14 @@ def returned_results(results):
                 "estimatedDiskBytes",
                 "expectedMemoryRssBytes",
                 "reclaimedMemoryBytes",
+                "outcome",
+                "configId",
+                "reclaimBytes",
+                "swappiness",
+                "cgroupMemoryDeltaBytes",
+                "hostMemAvailableDeltaBytes",
+                "kernelWriteAttempted",
+                "kernelWriteError",
                 "processCount",
                 "termSignalCount",
                 "killSignalCount",
@@ -1626,8 +1719,8 @@ def plan_payload(observed_at, preflight, protected, candidates, visible):
                 "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"],
+                "candidateKinds": ["cgroup-memory-reclaim", "browser-orphan-process-tree", "stale-web-observer-process-tree"],
+                "neverTouches": ["disk candidates", "apt cache", "journal", "tmp", "container images", "active observer process trees", "workload lifecycle"],
                 "termination": "SIGTERM, bounded wait, identity reclose, SIGKILL only same-tree residual",
             },
         }
@@ -1710,18 +1803,27 @@ def safe_unit_name(value):
         raw = "unidesk-%s-low-risk-gc" % re.sub(r"[^a-z0-9]+", "-", PROVIDER_ID.lower()).strip("-")
     return raw[:80]
 
+def required_policy_bool(key):
+    value = POLICY_TIMER_CONFIG.get(key) if isinstance(POLICY_TIMER_CONFIG, dict) else None
+    if not isinstance(value, bool):
+        raise RuntimeError("gc.remote.targets.%s.policyTimer.%s must be a YAML boolean" % (PROVIDER_ID, key))
+    return value
+
 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)
+    include_journal = required_policy_bool("includeJournal")
+    include_tmp = required_policy_bool("includeTmp")
+    include_apt_cache = required_policy_bool("includeAptCache")
+    include_tool_caches = required_policy_bool("includeToolCaches")
+    include_web_observe = required_policy_bool("includeWebObserveArtifacts")
+    include_k3s_images = required_policy_bool("includeK3sImageCache")
+    include_host_containerd = required_policy_bool("includeHostContainerdCache")
+    include_local_path_orphans = required_policy_bool("includeLocalPathOrphans")
+    include_kubernetes_retention = required_policy_bool("includeKubernetesObjectRetention")
     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
@@ -1733,6 +1835,9 @@ def render_remote_policy():
         "providerId": PROVIDER_ID,
         "unitName": unit_name,
         "statePath": state_path,
+        "includeJournal": include_journal,
+        "includeTmp": include_tmp,
+        "includeToolCaches": include_tool_caches,
         "journalTargetBytes": int(journal_target),
         "tmpMinAgeHours": tmp_min_age_hours,
         "tmpPrefixAllowlist": TMP_PREFIX_ALLOWLIST,
@@ -1743,18 +1848,23 @@ def render_remote_policy():
         "k3sImageCache": {"enabled": include_k3s_images, **CONTAINERD_CONFIG},
         "hostContainerdCache": {"enabled": include_host_containerd, **HOST_CONTAINERD_CONFIG},
         "localPathStorage": {"enabled": include_local_path_orphans, **LOCAL_PATH_CONFIG},
+        "kubernetesObjectRetention": (
+            {**KUBERNETES_RETENTION_CONFIG, "enabled": True}
+            if include_kubernetes_retention else {"enabled": False}
+        ),
         "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": "journal", "enabled": include_journal, "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": "tmp-allowlist", "enabled": include_tmp, "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"},
+        {"id": "kubernetes-object-retention", "enabled": include_kubernetes_retention, "risk": "medium", "mode": "terminal-owner-cascade-bounded"},
     ]
     service = "\n".join([
         "[Unit]",
@@ -1796,12 +1906,15 @@ def render_remote_policy():
         "journalTargetBytes": int(journal_target),
         "journalTarget": fmt_bytes(journal_target),
         "tmpMinAgeHours": tmp_min_age_hours,
+        "includeJournal": include_journal,
+        "includeTmp": include_tmp,
         "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,
+        "includeKubernetesObjectRetention": include_kubernetes_retention,
         "stages": stages,
         "policyConfig": policy_config,
         "script": POLICY_RUNNER_SOURCE,
@@ -1820,7 +1933,7 @@ def remote_policy_plan_payload(observed_at):
         "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"]},
+        "timer": {key: rendered.get(key) for key in ["unitName", "scriptPath", "configPath", "statePath", "servicePath", "timerPath", "onCalendar", "randomizedDelaySec", "journalTargetBytes", "journalTarget", "tmpMinAgeHours", "includeJournal", "includeTmp", "includeAptCache", "includeToolCaches", "includeWebObserveArtifacts", "includeK3sImageCache", "includeHostContainerdCache", "includeLocalPathOrphans", "includeKubernetesObjectRetention"]},
         "stages": rendered.get("stages"),
         "servicePreview": rendered["service"] if bool(OPTIONS.get("full")) else None,
         "timerPreview": rendered["timer"] if bool(OPTIONS.get("full")) else None,
@@ -1888,6 +2001,12 @@ def compact_policy_last_run(last):
                 "deletedPvcCount",
                 "protectedCount",
                 "candidateCount",
+                "requestedDeleteCount",
+                "deletedCount",
+                "objectCountDelta",
+                "configId",
+                "namespace",
+                "resource",
                 "reclaimedBytes",
             ]
             if item.get(key) is not None
@@ -1947,7 +2066,7 @@ def remote_policy_install_payload(observed_at):
         "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"]},
+        "timer": {key: rendered.get(key) for key in ["unitName", "scriptPath", "configPath", "statePath", "servicePath", "timerPath", "onCalendar", "randomizedDelaySec", "journalTargetBytes", "journalTarget", "tmpMinAgeHours", "includeJournal", "includeTmp", "includeAptCache", "includeToolCaches", "includeWebObserveArtifacts", "includeK3sImageCache", "includeHostContainerdCache", "includeLocalPathOrphans", "includeKubernetesObjectRetention"]},
         "stages": rendered.get("stages"),
         "systemd": {
             "daemonReload": bounded(daemon),
@@ -1956,6 +2075,247 @@ def remote_policy_install_payload(observed_at):
         },
     }
 
+def kubernetes_retention_plan_payload(observed_at):
+    try:
+        plan = kubernetes_retention_plan(
+            KUBERNETES_RETENTION_CONFIG,
+            "manual",
+            limit_override=int(OPTIONS.get("limit") or 1),
+            full=bool(OPTIONS.get("full")),
+        )
+    except Exception as exc:
+        return {
+            "ok": False,
+            "action": "gc remote retention plan",
+            "providerId": PROVIDER_ID,
+            "dryRun": True,
+            "mutation": False,
+            "observedAt": observed_at,
+            "error": "kubernetes-retention-plan-failed",
+            "message": str(exc),
+            "configSource": "config/unidesk-cli.yaml#gc.remote.targets.%s.kubernetesObjectRetention" % PROVIDER_ID,
+        }
+    plan = dict(plan)
+    plan.pop("selectedNames", None)
+    plan.update({
+        "action": "gc remote retention plan",
+        "providerId": PROVIDER_ID,
+        "observedAt": observed_at,
+        "configSource": "config/unidesk-cli.yaml#gc.remote.targets.%s.kubernetesObjectRetention" % PROVIDER_ID,
+        "runCommand": "bun scripts/cli.ts gc remote %s retention run --confirm --limit %s" % (
+            PROVIDER_ID,
+            int(OPTIONS.get("limit") or 1),
+        ),
+        "statusCommandTemplate": "bun scripts/cli.ts gc remote %s status --job-id <job-id>" % PROVIDER_ID,
+    })
+    return plan
+
+def update_kubernetes_retention_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("kubernetes retention 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_kubernetes_retention_job(job_id, observed_at, limit_override, worker_pid, worker_start_ticks):
+    paths = job_paths(job_id)
+    memory_before = collect_memory_snapshot()
+
+    def progress(completed_count, candidate_count):
+        update_kubernetes_retention_job_progress(
+            paths,
+            completed_count,
+            candidate_count,
+            worker_pid,
+            worker_start_ticks,
+        )
+
+    result = kubernetes_retention_execute(
+        KUBERNETES_RETENTION_CONFIG,
+        "manual",
+        limit_override=limit_override,
+        progress=progress,
+        full=bool(OPTIONS.get("full")),
+    )
+    memory_after = collect_memory_snapshot()
+    requested_count = safe_int(result.get("requestedDeleteCount"))
+    deleted_count = safe_int(result.get("deletedCount"))
+    payload = {
+        "ok": result.get("ok") is True,
+        "action": "gc remote retention run",
+        "providerId": PROVIDER_ID,
+        "scope": "kubernetes-object-retention",
+        "kind": "kubernetes-object-retention",
+        "status": "succeeded" if result.get("ok") is True else "failed",
+        "jobId": job_id,
+        "statusCommand": status_command(job_id),
+        "statePath": paths["state"],
+        "dryRun": False,
+        "mutation": result.get("mutation") is True,
+        "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": deleted_count, "candidateCount": requested_count},
+        "memoryBefore": memory_before,
+        "memoryAfter": memory_after,
+        "memoryAvailableDeltaBytes": (
+            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,
+        "summary": {
+            "configId": result.get("configId"),
+            "namespace": result.get("namespace"),
+            "resource": result.get("resource"),
+            "requestedDeleteCount": requested_count,
+            "deletedCount": deleted_count,
+            "objectCountDelta": result.get("objectCountDelta"),
+            "beforeCount": (result.get("before") or {}).get("count"),
+            "afterCount": (result.get("after") or {}).get("count"),
+            "failedBatchCount": len([item for item in result.get("batches") or [] if item.get("ok") is not True]),
+        },
+        "retention": result,
+        "configSource": "config/unidesk-cli.yaml#gc.remote.targets.%s.kubernetesObjectRetention" % PROVIDER_ID,
+    }
+    write_memory_job_state(paths, payload)
+
+def start_kubernetes_retention_job(observed_at):
+    try:
+        retention_config = kubernetes_retention_config(KUBERNETES_RETENTION_CONFIG)
+    except Exception as exc:
+        return {
+            "ok": False,
+            "action": "gc remote retention run",
+            "providerId": PROVIDER_ID,
+            "dryRun": True,
+            "mutation": False,
+            "status": "blocked",
+            "error": "kubernetes-retention-config-invalid",
+            "message": str(exc),
+        }
+    limit_override = int(OPTIONS.get("limit") or 1)
+    job_id = "%s-kubernetes-retention-%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)
+    base = {
+        "ok": True,
+        "action": "gc remote retention run",
+        "providerId": PROVIDER_ID,
+        "scope": "kubernetes-object-retention",
+        "kind": "kubernetes-object-retention",
+        "jobId": job_id,
+        "statusCommand": status_command(job_id),
+        "statePath": paths["state"],
+        "dryRun": False,
+        "mutation": False,
+        "mutationPending": True,
+        "observedAt": observed_at,
+        "startedAt": observed_at,
+        "configId": retention_config.get("configId"),
+        "namespace": retention_config.get("namespace"),
+        "resource": retention_config.get("resource"),
+        "limit": min(limit_override, int(retention_config.get("manualMaxDeletePerRun"))),
+        "configSource": "config/unidesk-cli.yaml#gc.remote.targets.%s.kubernetesObjectRetention" % PROVIDER_ID,
+    }
+    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", "error": "kubernetes-retention-fork-failed", "message": 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_kubernetes_retention_job(
+                    job_id,
+                    observed_at,
+                    limit_override,
+                    os.getpid(),
+                    process_start_ticks(os.getpid()),
+                )
+        except Exception as exc:
+            failed = dict(base)
+            failed.update({
+                "ok": False,
+                "status": "failed",
+                "error": "kubernetes-retention-worker-failed",
+                "message": str(exc),
+                "workerPid": os.getpid(),
+                "workerStartTicks": process_start_ticks(os.getpid()),
+                "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", "error": "kubernetes-retention-worker-identity-unavailable", "workerPid": child_pid, "finishedAt": now_iso()})
+        write_memory_job_state(paths, failed)
+        return failed
+    running = dict(base)
+    running.update({
+        "status": "running",
+        "workerPid": child_pid,
+        "workerStartTicks": worker_start_ticks,
+        "workerAlive": True,
+        "workerHeartbeatAt": now_iso(),
+        "progress": {"completedCount": 0, "candidateCount": None},
+    })
+    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", "error": "kubernetes-retention-worker-release-failed", "message": str(exc), "workerPid": child_pid, "workerStartTicks": worker_start_ticks, "finishedAt": now_iso()})
+        write_memory_job_state(paths, failed)
+        return failed
+    finally:
+        os.close(write_descriptor)
+    return running
+
 def update_memory_pressure_job_progress(paths, completed_count, candidate_count, worker_pid, worker_start_ticks):
     descriptor = acquire_job_lock(paths)
     try:
@@ -1983,8 +2343,15 @@ def finish_memory_pressure_job(job_id, candidates, observed_at, worker_pid, work
             item = dict(candidate)
             item.update({
                 "status": execution.get("status") or "succeeded",
+                "outcome": execution.get("outcome"),
                 "reclaimedBytes": 0,
                 "reclaimedMemoryBytes": execution.get("reclaimedMemoryBytes"),
+                "hostMemAvailableDeltaBytes": execution.get("hostMemAvailableDeltaBytes"),
+                "cgroupMemoryDeltaBytes": execution.get("cgroupMemoryDeltaBytes"),
+                "cgroupMemoryBeforeBytes": execution.get("cgroupMemoryBeforeBytes"),
+                "cgroupMemoryAfterBytes": execution.get("cgroupMemoryAfterBytes"),
+                "kernelWriteAttempted": execution.get("kernelWriteAttempted"),
+                "kernelWriteError": execution.get("kernelWriteError"),
                 "termSignalCount": execution.get("termSignalCount"),
                 "killSignalCount": execution.get("killSignalCount"),
                 "remainingIdentityCount": execution.get("remainingIdentityCount"),
@@ -2157,6 +2524,9 @@ def start_memory_pressure_job(observed_at, candidates):
 
 def main():
     observed_at = now_iso()
+    if ACTION == "memory-distribution":
+        emit_json(collect_memory_distribution(observed_at), persist_large=True)
+        return 0
     if ACTION == "memory":
         memory = collect_memory_snapshot()
         emit_json({
@@ -2170,6 +2540,7 @@ def main():
             "source": memory.get("source"),
             "swapExcluded": memory.get("swapExcluded") is True,
             "memory": memory.get("memory"),
+            "swap": memory.get("swap"),
             "error": memory.get("error"),
             "valuesRedacted": True,
         }, persist_large=False)
@@ -2177,6 +2548,12 @@ def main():
     if ACTION == "status" and job_id_or_none():
         emit_json(remote_gc_job_status(), persist_large=False)
         return 0
+    if ACTION == "retention-plan":
+        emit_json(kubernetes_retention_plan_payload(observed_at), persist_large=True)
+        return 0
+    if ACTION == "retention-run":
+        emit_json(start_kubernetes_retention_job(observed_at), 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)

scripts/src/gc-remote-web-observe.py

@@ -5,6 +5,8 @@ _OBSERVE_SCAN_ERROR = None
 _PROCESS_SCAN_ERROR = None
 _MEMORY_RECLAIM_CONFIG_ERROR = None
 _OBSERVE_ROOT_DIAGNOSTICS = {"configured": [], "readable": [], "missing": []}
+_PROCESS_PRESSURE_DIAGNOSTICS = {}
+_CGROUP_RECLAIM_DIAGNOSTICS = {"configuredCount": 0, "eligibleCount": 0, "protected": []}
 
 def configured_observe_roots():
     roots = config_list(MEMORY_CONFIG, "observeStateRoots", config_list(MEMORY_CONFIG, "webObserveRoots", []))
@@ -105,8 +107,61 @@ def assert_web_observe_candidate(path):
         raise RuntimeError("refusing to remove web-observe run with open fd/cwd reference: %s" % path)
     return record
 
+def cgroup_path_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 build_process_pressure_diagnostics(rows):
+    zombies = [row for row in rows.values() if row.get("state") == "Z"]
+    by_comm = {}
+    by_parent = {}
+    for row in zombies:
+        comm = str(row.get("comm") or "unknown")
+        by_comm[comm] = by_comm.get(comm, 0) + 1
+        ppid = safe_int(row.get("ppid"))
+        parent = rows.get(ppid) or {}
+        bucket = by_parent.setdefault(ppid, {
+            "ppid": ppid,
+            "parentComm": parent.get("comm"),
+            "parentCommandPreview": parent.get("commandPreview"),
+            "zombieCount": 0,
+        })
+        bucket["zombieCount"] += 1
+    top_rss = []
+    for row in sorted(rows.values(), key=lambda item: safe_int(item.get("rssBytes")), reverse=True)[:5]:
+        top_rss.append({
+            "pid": safe_int(row.get("pid")),
+            "ppid": safe_int(row.get("ppid")),
+            "comm": row.get("comm"),
+            "state": row.get("state"),
+            "rssBytes": safe_int(row.get("rssBytes")),
+            "rssHuman": fmt_bytes(safe_int(row.get("rssBytes"))),
+            "cgroupPath": cgroup_path_for_pid(row.get("pid")),
+        })
+    parent_rows = sorted(by_parent.values(), key=lambda item: safe_int(item.get("zombieCount")), reverse=True)[:5]
+    for item in parent_rows:
+        item.pop("parentCommandPreview", None)
+        item["cgroupPath"] = cgroup_path_for_pid(item.get("ppid")) if item.get("ppid") else None
+    return {
+        "processCount": len(rows),
+        "zombieCount": len(zombies),
+        "zombiesByComm": [
+            {"comm": comm, "zombieCount": count}
+            for comm, count in sorted(by_comm.items(), key=lambda item: item[1], reverse=True)[:5]
+        ],
+        "zombiesByParent": parent_rows,
+        "topRss": top_rss,
+    }
+
 def read_proc_process_table():
-    global _PROCESS_SCAN_TRUNCATED, _PROCESS_SCAN_ERROR
+    global _PROCESS_SCAN_TRUNCATED, _PROCESS_SCAN_ERROR, _PROCESS_PRESSURE_DIAGNOSTICS
     scan_limit = config_int(MEMORY_CONFIG, "processScanLimit", 16384, minimum=1, maximum=1000000)
     try:
         clock_ticks = int(os.sysconf("SC_CLK_TCK"))
@@ -129,10 +184,12 @@ def read_proc_process_table():
             close = stat_text.rfind(")")
             if close < 0:
                 continue
+            comm = stat_text[stat_text.find("(") + 1:close]
             fields = stat_text[close + 2:].split()
             if len(fields) < 22:
                 continue
             pid = int(pid_name)
+            state = fields[0]
             ppid = int(fields[1])
             sid = int(fields[3])
             start_ticks = int(fields[19])
@@ -150,6 +207,8 @@ def read_proc_process_table():
                 "pid": pid,
                 "ppid": ppid,
                 "sid": sid,
+                "comm": comm,
+                "state": state,
                 "startTicks": start_ticks,
                 "ageSeconds": max(0, int(uptime - (float(start_ticks) / float(clock_ticks)))),
                 "rssBytes": rss_bytes,
@@ -163,6 +222,7 @@ def read_proc_process_table():
         _PROCESS_SCAN_ERROR = "proc-process-table-empty"
     elif stat_read_failures > max(32, len(pid_names) // 2):
         _PROCESS_SCAN_ERROR = "proc-process-stat-read-failure-threshold:%s-of-%s" % (stat_read_failures, len(pid_names))
+    _PROCESS_PRESSURE_DIAGNOSTICS = build_process_pressure_diagnostics(rows)
     return rows
 
 def validate_observe_run_markers(path):
@@ -250,14 +310,157 @@ def command_matches(row, patterns):
 def memory_process_candidate_id(kind, root):
     return "%s:%s:%s:%s" % (kind, safe_int(root.get("sid")), safe_int(root.get("pid")), safe_int(root.get("startTicks")))
 
+def read_cgroup_integer(path, name):
+    try:
+        with open(os.path.join(path, name), "r", encoding="utf-8") as handle:
+            return int(handle.read().strip())
+    except (OSError, ValueError):
+        return None
+
+def read_cgroup_memory_stat(path):
+    selected = {}
+    try:
+        with open(os.path.join(path, "memory.stat"), "r", encoding="utf-8") as handle:
+            for line in handle:
+                parts = line.split()
+                if len(parts) == 2 and parts[0] in set(["anon", "file", "inactive_anon", "inactive_file"]):
+                    selected[parts[0]] = safe_int(parts[1])
+    except OSError:
+        return {}
+    return selected
+
+def configured_cgroup_reclaim_targets():
+    global _MEMORY_RECLAIM_CONFIG_ERROR
+    policy = MEMORY_CONFIG.get("cgroupReclaim")
+    if policy is None:
+        return []
+    if not isinstance(policy, dict):
+        _MEMORY_RECLAIM_CONFIG_ERROR = "yaml-cgroup-reclaim-policy-invalid"
+        return []
+    if policy.get("enabled") is not True:
+        return []
+    target_available = policy.get("targetMemAvailableBytes")
+    swap_reserve = policy.get("minimumSwapFreeBytesAfterReclaim")
+    targets = policy.get("targets")
+    if isinstance(target_available, bool) or not isinstance(target_available, int) or target_available <= 0:
+        _MEMORY_RECLAIM_CONFIG_ERROR = "yaml-cgroup-reclaim-target-memory-invalid"
+        return []
+    if isinstance(swap_reserve, bool) or not isinstance(swap_reserve, int) or swap_reserve < 0:
+        _MEMORY_RECLAIM_CONFIG_ERROR = "yaml-cgroup-reclaim-swap-reserve-invalid"
+        return []
+    if not isinstance(targets, list) or not targets:
+        _MEMORY_RECLAIM_CONFIG_ERROR = "yaml-cgroup-reclaim-targets-invalid"
+        return []
+    validated = []
+    seen_ids = set()
+    for index, item in enumerate(targets):
+        if not isinstance(item, dict):
+            _MEMORY_RECLAIM_CONFIG_ERROR = "yaml-cgroup-reclaim-target-invalid:%s" % index
+            return []
+        target_id = item.get("id")
+        path = item.get("path")
+        reclaim_bytes = item.get("reclaimBytes")
+        swappiness = item.get("swappiness")
+        if not isinstance(target_id, str) or not re.match(r"^[A-Za-z0-9._-]{1,80}$", target_id) or target_id in seen_ids:
+            _MEMORY_RECLAIM_CONFIG_ERROR = "yaml-cgroup-reclaim-target-id-invalid:%s" % index
+            return []
+        if not isinstance(path, str) or (path != "/sys/fs/cgroup" and not path.startswith("/sys/fs/cgroup/")) or os.path.normpath(path) != path:
+            _MEMORY_RECLAIM_CONFIG_ERROR = "yaml-cgroup-reclaim-target-path-invalid:%s" % target_id
+            return []
+        if isinstance(reclaim_bytes, bool) or not isinstance(reclaim_bytes, int) or reclaim_bytes <= 0 or reclaim_bytes > 8 * 1024 * 1024 * 1024:
+            _MEMORY_RECLAIM_CONFIG_ERROR = "yaml-cgroup-reclaim-bytes-invalid:%s" % target_id
+            return []
+        if isinstance(swappiness, bool) or not isinstance(swappiness, int) or swappiness < 0 or swappiness > 200:
+            _MEMORY_RECLAIM_CONFIG_ERROR = "yaml-cgroup-reclaim-swappiness-invalid:%s" % target_id
+            return []
+        priority = item.get("priority", index + 1)
+        if isinstance(priority, bool) or not isinstance(priority, int) or priority < 1 or priority > 1000:
+            _MEMORY_RECLAIM_CONFIG_ERROR = "yaml-cgroup-reclaim-priority-invalid:%s" % target_id
+            return []
+        seen_ids.add(target_id)
+        validated.append({
+            "id": target_id,
+            "path": path,
+            "reclaimBytes": reclaim_bytes,
+            "swappiness": swappiness,
+            "priority": priority,
+            "targetMemAvailableBytes": target_available,
+            "minimumSwapFreeBytesAfterReclaim": swap_reserve,
+        })
+    return sorted(validated, key=lambda item: item.get("priority"))
+
+def collect_cgroup_reclaim_candidates():
+    global _MEMORY_RECLAIM_CONFIG_ERROR, _CGROUP_RECLAIM_DIAGNOSTICS
+    targets = configured_cgroup_reclaim_targets()
+    diagnostics = {"configuredCount": len(targets), "eligibleCount": 0, "protected": []}
+    _CGROUP_RECLAIM_DIAGNOSTICS = diagnostics
+    if _MEMORY_RECLAIM_CONFIG_ERROR or not targets:
+        return []
+    host = collect_memory_snapshot()
+    if host.get("ok") is not True:
+        _MEMORY_RECLAIM_CONFIG_ERROR = "host-memory-snapshot-unavailable"
+        return []
+    available = safe_int((host.get("memory") or {}).get("availableBytes"))
+    swap_free = safe_int((host.get("swap") or {}).get("freeBytes"))
+    candidates = []
+    for item in targets:
+        path = item.get("path")
+        reclaim_path = os.path.join(path, "memory.reclaim")
+        protected_reason = None
+        if os.path.realpath(path) != path or not os.path.isdir(path) or os.path.islink(path):
+            protected_reason = "cgroup-path-unavailable-or-redirected"
+        elif not os.path.exists(reclaim_path) or not os.access(reclaim_path, os.W_OK):
+            protected_reason = "memory-reclaim-unavailable"
+        elif available > safe_int(item.get("targetMemAvailableBytes")):
+            protected_reason = "target-memavailable-already-met"
+        elif safe_int(item.get("swappiness")) > 0 and swap_free - safe_int(item.get("reclaimBytes")) < safe_int(item.get("minimumSwapFreeBytesAfterReclaim")):
+            protected_reason = "minimum-swap-free-reserve"
+        current = read_cgroup_integer(path, "memory.current")
+        if protected_reason is not None:
+            diagnostics["protected"].append({
+                "id": item.get("id"),
+                "path": path,
+                "reason": protected_reason,
+                "memoryCurrentBytes": current,
+            })
+            continue
+        memory_stat = read_cgroup_memory_stat(path)
+        reclaim_bytes = safe_int(item.get("reclaimBytes"))
+        candidates.append({
+            "id": "cgroup-memory-reclaim:%s" % item.get("id"),
+            "configId": item.get("id"),
+            "kind": "cgroup-memory-reclaim",
+            "risk": "medium",
+            "description": "Request bounded cgroup v2 memory reclaim from an exact YAML allowlist path",
+            "reason": "host-memavailable-below-yaml-target",
+            "path": path,
+            "priority": safe_int(item.get("priority")),
+            "reclaimBytes": reclaim_bytes,
+            "swappiness": safe_int(item.get("swappiness")),
+            "targetMemAvailableBytes": safe_int(item.get("targetMemAvailableBytes")),
+            "minimumSwapFreeBytesAfterReclaim": safe_int(item.get("minimumSwapFreeBytesAfterReclaim")),
+            "memoryCurrentBytes": current,
+            "memorySwapCurrentBytes": read_cgroup_integer(path, "memory.swap.current"),
+            "memoryStat": memory_stat,
+            "expectedMemoryRssBytes": reclaim_bytes,
+            "expectedMemoryRss": fmt_bytes(reclaim_bytes),
+            "estimatedDiskBytes": 0,
+            "estimatedReclaimBytes": 0,
+            "action": {"op": "cgroup-v2-memory-reclaim", "allowlist": "yaml-exact-cgroup-path"},
+        })
+    diagnostics["eligibleCount"] = len(candidates)
+    return candidates
+
 def collect_memory_reclaim_candidates():
-    global _OBSERVE_SCAN_TRUNCATED, _PROCESS_SCAN_TRUNCATED, _OBSERVE_SCAN_ERROR, _PROCESS_SCAN_ERROR, _MEMORY_RECLAIM_CONFIG_ERROR, _OBSERVE_ROOT_DIAGNOSTICS
+    global _OBSERVE_SCAN_TRUNCATED, _PROCESS_SCAN_TRUNCATED, _OBSERVE_SCAN_ERROR, _PROCESS_SCAN_ERROR, _MEMORY_RECLAIM_CONFIG_ERROR, _OBSERVE_ROOT_DIAGNOSTICS, _PROCESS_PRESSURE_DIAGNOSTICS, _CGROUP_RECLAIM_DIAGNOSTICS
     _OBSERVE_SCAN_TRUNCATED = False
     _PROCESS_SCAN_TRUNCATED = False
     _OBSERVE_SCAN_ERROR = None
     _PROCESS_SCAN_ERROR = None
     _MEMORY_RECLAIM_CONFIG_ERROR = None
     _OBSERVE_ROOT_DIAGNOSTICS = {"configured": [], "readable": [], "missing": []}
+    _PROCESS_PRESSURE_DIAGNOSTICS = {}
+    _CGROUP_RECLAIM_DIAGNOSTICS = {"configuredCount": 0, "eligibleCount": 0, "protected": []}
     _OBSERVE_RUN_CACHE.clear()
     required_keys = [
         "observeStateRoots", "minimumReadableObserveRoots", "observeRunDepth", "observeScanLimit", "staleRunMaxAgeHours",
@@ -329,7 +532,7 @@ def collect_memory_reclaim_candidates():
     protected_pids = set()
     for pid in fresh_observer_pids:
         protected_pids.update(process_descendants(processes, pid))
-    candidates = []
+    candidates = collect_cgroup_reclaim_candidates()
     for sid in sorted(set(safe_int(row.get("sid")) for row in processes.values() if safe_int(row.get("sid")) > 0)):
         closed = session_closed_tree(processes, sid)
         if closed is None:
@@ -386,7 +589,11 @@ def collect_memory_reclaim_candidates():
             "action": {"op": "term-wait-kill-exact-process-tree", "allowlist": "yaml-web-probe-memory-pressure"},
         })
     max_candidates = config_int(MEMORY_CONFIG, "maxMemoryReclaimCandidates", 20, minimum=1, maximum=100)
-    return sorted(candidates, key=lambda item: safe_int(item.get("expectedMemoryRssBytes")), reverse=True)[:max_candidates]
+    return sorted(candidates, key=lambda item: (
+        0 if item.get("kind") == "cgroup-memory-reclaim" else 1,
+        safe_int(item.get("priority"), 1000),
+        -safe_int(item.get("expectedMemoryRssBytes")),
+    ))[:max_candidates]
 
 def memory_reclaim_scan_diagnostics():
     blocked = bool(_MEMORY_RECLAIM_CONFIG_ERROR or _OBSERVE_SCAN_TRUNCATED or _PROCESS_SCAN_TRUNCATED or _OBSERVE_SCAN_ERROR or _PROCESS_SCAN_ERROR)
@@ -398,6 +605,8 @@ def memory_reclaim_scan_diagnostics():
         "processScanTruncated": bool(_PROCESS_SCAN_TRUNCATED),
         "observeScanError": _OBSERVE_SCAN_ERROR,
         "processScanError": _PROCESS_SCAN_ERROR,
+        "processPressure": _PROCESS_PRESSURE_DIAGNOSTICS,
+        "cgroupReclaim": _CGROUP_RECLAIM_DIAGNOSTICS,
         "observeRoots": {
             "configuredCount": len(_OBSERVE_ROOT_DIAGNOSTICS.get("configured") or []),
             "readableCount": len(_OBSERVE_ROOT_DIAGNOSTICS.get("readable") or []),
@@ -453,6 +662,70 @@ def extend_with_allowlisted_session_members(identities, processes, sid, browser_
         known.add(identity_key)
     return added
 
+def execute_cgroup_reclaim_candidate(candidate):
+    config_id = candidate.get("configId")
+    configured = next((item for item in configured_cgroup_reclaim_targets() if item.get("id") == config_id), None)
+    if configured is None:
+        raise RuntimeError("cgroup reclaim candidate is no longer present in YAML allowlist")
+    for key in ["path", "reclaimBytes", "swappiness", "targetMemAvailableBytes", "minimumSwapFreeBytesAfterReclaim"]:
+        if configured.get(key) != candidate.get(key):
+            raise RuntimeError("cgroup reclaim candidate no longer matches YAML: %s" % key)
+    path = configured.get("path")
+    reclaim_path = os.path.join(path, "memory.reclaim")
+    if os.path.realpath(path) != path or not os.path.isdir(path) or os.path.islink(path):
+        raise RuntimeError("cgroup reclaim path is unavailable or redirected")
+    if not os.path.exists(reclaim_path) or not os.access(reclaim_path, os.W_OK):
+        raise RuntimeError("cgroup memory.reclaim is unavailable")
+    before = collect_memory_snapshot()
+    before_available = safe_int((before.get("memory") or {}).get("availableBytes"), None)
+    if before_available is None:
+        raise RuntimeError("MemAvailable is unavailable before cgroup reclaim")
+    if before_available > safe_int(configured.get("targetMemAvailableBytes")):
+        return {
+            "status": "succeeded",
+            "outcome": "target-memavailable-already-met",
+            "reclaimedMemoryBytes": 0,
+            "memoryBefore": before,
+            "memoryAfter": before,
+            "cgroupMemoryBeforeBytes": read_cgroup_integer(path, "memory.current"),
+            "cgroupMemoryAfterBytes": read_cgroup_integer(path, "memory.current"),
+            "kernelWriteAttempted": False,
+        }
+    swap_free = safe_int((before.get("swap") or {}).get("freeBytes"), None)
+    if safe_int(configured.get("swappiness")) > 0:
+        if swap_free is None:
+            raise RuntimeError("SwapFree is unavailable before cgroup reclaim")
+        required = safe_int(configured.get("reclaimBytes")) + safe_int(configured.get("minimumSwapFreeBytesAfterReclaim"))
+        if swap_free < required:
+            raise RuntimeError("cgroup reclaim would violate YAML minimum SwapFree reserve")
+    cgroup_before = read_cgroup_integer(path, "memory.current")
+    write_error = None
+    try:
+        with open(reclaim_path, "w", encoding="utf-8") as handle:
+            handle.write("%s swappiness=%s\n" % (safe_int(configured.get("reclaimBytes")), safe_int(configured.get("swappiness"))))
+    except OSError as exc:
+        write_error = "%s:%s" % (type(exc).__name__, getattr(exc, "errno", None))
+    time.sleep(0.2)
+    after = collect_memory_snapshot()
+    after_available = safe_int((after.get("memory") or {}).get("availableBytes"), None)
+    cgroup_after = read_cgroup_integer(path, "memory.current")
+    cgroup_delta = max(0, cgroup_before - cgroup_after) if cgroup_before is not None and cgroup_after is not None else 0
+    host_delta = max(0, after_available - before_available) if after_available is not None else 0
+    succeeded = write_error is None or cgroup_delta > 0 or host_delta > 0
+    return {
+        "status": "succeeded" if succeeded else "failed",
+        "outcome": "partial-reclaim-after-kernel-error" if write_error and succeeded else ("reclaim-requested" if succeeded else "reclaim-request-failed"),
+        "reclaimedMemoryBytes": max(cgroup_delta, host_delta),
+        "hostMemAvailableDeltaBytes": (after_available - before_available) if after_available is not None else None,
+        "cgroupMemoryDeltaBytes": cgroup_delta,
+        "cgroupMemoryBeforeBytes": cgroup_before,
+        "cgroupMemoryAfterBytes": cgroup_after,
+        "kernelWriteAttempted": True,
+        "kernelWriteError": write_error,
+        "memoryBefore": before,
+        "memoryAfter": after,
+    }
+
 def execute_memory_process_candidate(candidate):
     current = next((item for item in collect_memory_reclaim_candidates() if item.get("id") == candidate.get("id")), None)
     if current is None:

scripts/src/gc-remote.ts

@@ -5,7 +5,7 @@ import { type UniDeskConfig, rootPath } from "./config";
 import { remoteGcDegradedFailure } from "./gc-remote-degraded";
 import { runSshCommandCapture } from "./ssh";
 
-type RemoteGcAction = "plan" | "memory" | "snapshot" | "trend" | "run" | "status" | "policy-plan" | "policy-install" | "policy-status";
+type RemoteGcAction = "plan" | "memory" | "memory-distribution" | "retention-plan" | "retention-run" | "snapshot" | "trend" | "run" | "status" | "policy-plan" | "policy-install" | "policy-status";
 
 interface RemoteGcOptions {
   confirm: boolean;
@@ -75,6 +75,11 @@ const GC_REMOTE_RUNNER_RELATIVE_PATH = "scripts/src/gc-remote-runner.py";
 const GC_REMOTE_RUNNER_CONFIG_PLACEHOLDER = "__UNIDESK_GC_REMOTE_CONFIG_BASE64__";
 const GC_REMOTE_WEB_OBSERVE_RELATIVE_PATH = "scripts/src/gc-remote-web-observe.py";
 const GC_REMOTE_WEB_OBSERVE_PLACEHOLDER = "# __UNIDESK_GC_REMOTE_WEB_OBSERVE_HELPERS__";
+const GC_REMOTE_MEMORY_DISTRIBUTION_RELATIVE_PATH = "scripts/src/gc-remote-memory-distribution.py";
+const GC_REMOTE_MEMORY_DISTRIBUTION_PLACEHOLDER = "# __UNIDESK_GC_REMOTE_MEMORY_DISTRIBUTION_HELPERS__";
+const GC_REMOTE_KUBERNETES_RETENTION_RELATIVE_PATH = "scripts/src/gc-remote-kubernetes-retention.py";
+const GC_REMOTE_KUBERNETES_RETENTION_PLACEHOLDER = "# __UNIDESK_GC_REMOTE_KUBERNETES_RETENTION_HELPERS__";
+const GC_REMOTE_POLICY_KUBERNETES_RETENTION_PLACEHOLDER = "# __UNIDESK_GC_REMOTE_POLICY_KUBERNETES_RETENTION_HELPERS__";
 const GC_REMOTE_CONTAINERD_RELATIVE_PATH = "scripts/src/gc-remote-containerd.py";
 const GC_REMOTE_CONTAINERD_PLACEHOLDER = "# __UNIDESK_GC_REMOTE_CONTAINERD_HELPERS__";
 const GC_REMOTE_PVC_RELATIVE_PATH = "scripts/src/gc-remote-pvc.py";
@@ -91,10 +96,40 @@ export async function runRemoteGcCommand(config: UniDeskConfig, providerId: stri
     return {
       ok: false,
       error: "gc-remote-provider-required",
-      usage: "bun scripts/cli.ts gc remote <providerId> memory|plan|snapshot|trend|run|status|policy [--confirm]",
+      usage: "bun scripts/cli.ts gc remote <providerId> memory|memory-distribution|retention|plan|snapshot|trend|run|status|policy [--confirm]",
     };
   }
   const subaction = action ?? "plan";
+  if (subaction === "retention") {
+    const [retentionAction = "plan", ...retentionArgs] = args;
+    const options = parseRemoteGcOptions(retentionArgs);
+    if (retentionAction === "plan" || retentionAction === "dry-run") {
+      return await runRemoteGc(config, providerId, "retention-plan", options);
+    }
+    if (retentionAction === "run") {
+      if (!options.confirm) {
+        return {
+          ok: false,
+          error: "gc-remote-retention-run-requires-confirm",
+          dryRun: true,
+          mutation: false,
+          requiredFlag: "--confirm",
+          planCommand: `bun scripts/cli.ts gc remote ${providerId} retention plan --limit ${options.limit}`,
+          runCommand: `bun scripts/cli.ts gc remote ${providerId} retention run --confirm --limit ${options.limit}`,
+        };
+      }
+      return await runRemoteGc(config, providerId, "retention-run", options);
+    }
+    if (retentionAction === "status") {
+      return await runRemoteGc(config, providerId, "status", options);
+    }
+    return {
+      ok: false,
+      error: "unsupported-gc-remote-retention-action",
+      action: retentionAction,
+      supportedActions: ["plan", "run", "status"],
+    };
+  }
   if (subaction === "policy") {
     const [policyAction = "plan", ...policyArgs] = args;
     const options = parseRemoteGcOptions(policyArgs);
@@ -137,6 +172,9 @@ export async function runRemoteGcCommand(config: UniDeskConfig, providerId: stri
     };
   }
   if (subaction === "memory") return await runRemoteGc(config, providerId, "memory", options);
+  if (subaction === "memory-distribution" || subaction === "memory-overview") {
+    return await runRemoteGc(config, providerId, "memory-distribution", options);
+  }
   if (subaction === "plan" || subaction === "dry-run") return await runRemoteGc(config, providerId, "plan", options);
   if (subaction === "snapshot" || subaction === "growth") return await runRemoteGc(config, providerId, "snapshot", options);
   if (subaction === "trend") return await runRemoteGc(config, providerId, "trend", options);
@@ -159,7 +197,7 @@ export async function runRemoteGcCommand(config: UniDeskConfig, providerId: stri
     ok: false,
     error: "unsupported-gc-remote-action",
     action: subaction,
-    supportedActions: ["memory", "plan", "snapshot", "trend", "run", "status", "policy"],
+    supportedActions: ["memory", "memory-distribution", "retention", "plan", "snapshot", "trend", "run", "status", "policy"],
   };
 }
 
@@ -352,6 +390,12 @@ function remoteGcPython(configBase64: string): string {
   if (!template.includes(GC_REMOTE_WEB_OBSERVE_PLACEHOLDER)) {
     throw new Error(`${GC_REMOTE_RUNNER_RELATIVE_PATH} missing ${GC_REMOTE_WEB_OBSERVE_PLACEHOLDER}`);
   }
+  if (!template.includes(GC_REMOTE_MEMORY_DISTRIBUTION_PLACEHOLDER)) {
+    throw new Error(`${GC_REMOTE_RUNNER_RELATIVE_PATH} missing ${GC_REMOTE_MEMORY_DISTRIBUTION_PLACEHOLDER}`);
+  }
+  if (!template.includes(GC_REMOTE_KUBERNETES_RETENTION_PLACEHOLDER)) {
+    throw new Error(`${GC_REMOTE_RUNNER_RELATIVE_PATH} missing ${GC_REMOTE_KUBERNETES_RETENTION_PLACEHOLDER}`);
+  }
   if (!template.includes(GC_REMOTE_CONTAINERD_PLACEHOLDER)) {
     throw new Error(`${GC_REMOTE_RUNNER_RELATIVE_PATH} missing ${GC_REMOTE_CONTAINERD_PLACEHOLDER}`);
   }
@@ -368,17 +412,28 @@ function remoteGcPython(configBase64: string): string {
     throw new Error(`${GC_REMOTE_RUNNER_RELATIVE_PATH} missing ${GC_REMOTE_POLICY_RUNNER_PLACEHOLDER}`);
   }
   const webObserveHelpers = readFileSync(rootPath(GC_REMOTE_WEB_OBSERVE_RELATIVE_PATH), "utf8");
+  const memoryDistributionHelpers = readFileSync(rootPath(GC_REMOTE_MEMORY_DISTRIBUTION_RELATIVE_PATH), "utf8");
+  const kubernetesRetentionHelpers = readFileSync(rootPath(GC_REMOTE_KUBERNETES_RETENTION_RELATIVE_PATH), "utf8");
   const containerdHelpers = readFileSync(rootPath(GC_REMOTE_CONTAINERD_RELATIVE_PATH), "utf8");
   const pvcHelpers = readFileSync(rootPath(GC_REMOTE_PVC_RELATIVE_PATH), "utf8");
   const growthHelpers = readFileSync(rootPath(GC_REMOTE_GROWTH_RELATIVE_PATH), "utf8");
   const registryHelpers = readFileSync(rootPath(GC_REMOTE_REGISTRY_RELATIVE_PATH), "utf8");
-  const policyRunner = readFileSync(rootPath(GC_REMOTE_POLICY_RUNNER_RELATIVE_PATH), "utf8");
+  const policyRunnerTemplate = readFileSync(rootPath(GC_REMOTE_POLICY_RUNNER_RELATIVE_PATH), "utf8");
+  if (!policyRunnerTemplate.includes(GC_REMOTE_POLICY_KUBERNETES_RETENTION_PLACEHOLDER)) {
+    throw new Error(`${GC_REMOTE_POLICY_RUNNER_RELATIVE_PATH} missing ${GC_REMOTE_POLICY_KUBERNETES_RETENTION_PLACEHOLDER}`);
+  }
+  const policyRunner = policyRunnerTemplate.replace(
+    GC_REMOTE_POLICY_KUBERNETES_RETENTION_PLACEHOLDER,
+    () => kubernetesRetentionHelpers.trimEnd(),
+  );
   return template
-    .replace(GC_REMOTE_WEB_OBSERVE_PLACEHOLDER, webObserveHelpers.trimEnd())
-    .replace(GC_REMOTE_CONTAINERD_PLACEHOLDER, containerdHelpers.trimEnd())
-    .replace(GC_REMOTE_PVC_PLACEHOLDER, pvcHelpers.trimEnd())
-    .replace(GC_REMOTE_GROWTH_PLACEHOLDER, growthHelpers.trimEnd())
-    .replace(GC_REMOTE_REGISTRY_PLACEHOLDER, registryHelpers.trimEnd())
+    .replace(GC_REMOTE_MEMORY_DISTRIBUTION_PLACEHOLDER, () => memoryDistributionHelpers.trimEnd())
+    .replace(GC_REMOTE_KUBERNETES_RETENTION_PLACEHOLDER, () => kubernetesRetentionHelpers.trimEnd())
+    .replace(GC_REMOTE_WEB_OBSERVE_PLACEHOLDER, () => webObserveHelpers.trimEnd())
+    .replace(GC_REMOTE_CONTAINERD_PLACEHOLDER, () => containerdHelpers.trimEnd())
+    .replace(GC_REMOTE_PVC_PLACEHOLDER, () => pvcHelpers.trimEnd())
+    .replace(GC_REMOTE_GROWTH_PLACEHOLDER, () => growthHelpers.trimEnd())
+    .replace(GC_REMOTE_REGISTRY_PLACEHOLDER, () => registryHelpers.trimEnd())
     .replace(GC_REMOTE_POLICY_RUNNER_PLACEHOLDER, Buffer.from(policyRunner, "utf8").toString("base64"))
     .replace(GC_REMOTE_RUNNER_CONFIG_PLACEHOLDER, configBase64);
 }

scripts/src/help.ts

@@ -519,6 +519,9 @@ function gcHelp(): unknown {
       "bun scripts/cli.ts gc policy install",
       "bun scripts/cli.ts gc remote G14 plan",
       "bun scripts/cli.ts gc remote NC01 memory",
+      "bun scripts/cli.ts gc remote NC01 memory-distribution",
+      "bun scripts/cli.ts gc remote NC01 retention plan --limit 500",
+      "bun scripts/cli.ts gc remote NC01 retention run --confirm --limit 500",
       "bun scripts/cli.ts gc remote NC01 plan --memory-pressure-only",
       "bun scripts/cli.ts gc remote NC01 run --confirm --memory-pressure-only",
       "bun scripts/cli.ts gc remote NC01 status --job-id <job-id>",
@@ -569,6 +572,8 @@ function gcHelp(): unknown {
       "policy plan|install": "render or install journald caps and a daily low-risk gc systemd timer from config/unidesk-cli.yaml#gc.policyTimer",
       "remote <providerId> plan|run": "run bounded GC through UniDesk SSH passthrough on a provider host; G14 protects HWLAB k3s/runtime/PVC/workspace paths, and HWLAB registry retention is explicit opt-in with workload-ref, digest-closure, recent-tag and per-repo tag protection",
       "remote <providerId> memory": "只读输出目标节点 /proc/meminfo 的 MemAvailable；不合并 swap、cgroup 或 virtual memory，也不执行 GC mutation",
+      "remote <providerId> memory-distribution": "只读输出目标节点物理内存、PSS、cgroup、zombie 和 Kubernetes 对象分布的紧凑 overview",
+      "remote <providerId> retention plan|run|status": "按 owning YAML 对终态 Kubernetes 对象执行 fail-closed 计划、异步确认回收和状态查询；自动 policy 复用同一筛选器",
       "--no-file-logs|--no-docker-logs|--no-journal|--no-build-cache|--no-tmp|--no-db-summary": "disable one collector",
     },
     reference: "docs/reference/gc.md",

src/components/microservices/decision-center/Dockerfile

@@ -3,7 +3,7 @@ FROM ${DECISION_CENTER_BASE_IMAGE}
 
 WORKDIR /app/src/components/microservices/decision-center
 RUN apt-get update \
-  && apt-get install -y --no-install-recommends git openssh-client ca-certificates \
+  && apt-get install -y --no-install-recommends git openssh-client ca-certificates tini \
   && apt-get clean \
   && rm -rf /var/lib/apt/lists/*
 COPY src/components/microservices/decision-center/package.json ./package.json
@@ -13,4 +13,5 @@ COPY src/components/shared /app/src/components/shared
 COPY src/components/microservices/decision-center/src ./src
 
 EXPOSE 4277
+ENTRYPOINT ["tini", "--"]
 CMD ["bun", "run", "src/index.ts"]

src/components/microservices/decision-center/src/index.ts

@@ -38,6 +38,7 @@ interface RuntimeConfig {
   databaseUrl: string;
   logFile: string;
   databasePoolMax: number;
+  storageHealthRefreshMs: number;
   storage: StorageConfig;
 }
 
@@ -196,6 +197,15 @@ function envNumber(name: string, fallback: number): number {
   return Number.isFinite(value) && value > 0 ? Math.floor(value) : fallback;
 }
 
+function requiredEnvNumber(name: string): number {
+  const raw = process.env[name];
+  const value = Number(raw);
+  if (raw === undefined || raw.length === 0 || !Number.isFinite(value) || value <= 0) {
+    throw new Error(`${name} must be a positive number`);
+  }
+  return Math.floor(value);
+}
+
 function configFromEnv(): RuntimeConfig {
   const databaseUrl = process.env.DATABASE_URL || (isCheckMode ? "postgres://syntax-check@127.0.0.1:1/syntax-check" : "");
   if (!databaseUrl) throw new Error("DATABASE_URL is required");
@@ -207,6 +217,7 @@ function configFromEnv(): RuntimeConfig {
     databaseUrl,
     logFile: envString("LOG_FILE", "/var/log/unidesk/decision-center.jsonl"),
     databasePoolMax: Math.max(1, Math.min(8, envNumber("DATABASE_POOL_MAX", 2))),
+    storageHealthRefreshMs: requiredEnvNumber("DECISION_CENTER_STORAGE_HEALTH_REFRESH_MS"),
     storage: {
       primary: storagePrimary,
       repoSshUrl: envString("DECISION_CENTER_GIT_REPO_SSH_URL", ""),
@@ -2188,8 +2199,15 @@ if (import.meta.main && !isCheckMode) {
   setInterval(() => {
     if (schemaReady) {
       void refreshDatabaseHealth();
-      void updateStorageHealth();
     }
   }, 15_000).unref();
-  log("info", "service_started", { host: config.host, port: config.port, storage: storageConfigSummary() });
+  setInterval(() => {
+    if (schemaReady) void updateStorageHealth();
+  }, config.storageHealthRefreshMs).unref();
+  log("info", "service_started", {
+    host: config.host,
+    port: config.port,
+    storage: storageConfigSummary(),
+    storageHealthRefreshMs: config.storageHealthRefreshMs,
+  });
 }