def registry_growth_snapshot():
    summary = {
        "path": REGISTRY_ROOT,
        "sizeBytes": du_size(REGISTRY_ROOT, 60) or 0,
    }
    summary["sizeHuman"] = fmt_bytes(summary["sizeBytes"])
    if OPTIONS.get("hwlabRegistry", False):
        plan = plan_registry_retention()
        retention = dict(plan.get("summary") or {})
        for key in ["registrySizeBytes", "estimatedReclaimBytes"]:
            if key in retention:
                retention[key.replace("Bytes", "Human")] = fmt_bytes(retention.get(key) or 0)
        summary["retentionPlan"] = retention
    else:
        summary["retentionPlan"] = {
            "skipped": True,
            "reason": "rerun snapshot with --include-hwlab-registry to compute tag/revision retention counters",
        }
    summary["cadence"] = {
        "dryRun": "daily or before/after every v0.2 CI/CD burst",
        "maintenanceRun": "weekly, or when root >=80%, or when registry growth exceeds the agreed daily threshold",
        "planCommand": "bun scripts/cli.ts gc remote %s plan --target-use-percent 70 --include-hwlab-registry --limit 50" % PROVIDER_ID,
        "snapshotCommand": "bun scripts/cli.ts gc remote %s snapshot --include-hwlab-registry --history-limit 12" % PROVIDER_ID,
        "runCommand": "bun scripts/cli.ts gc remote %s run --confirm --include-hwlab-registry --target-use-percent 70 --limit 50" % PROVIDER_ID,
        "defaultRetention": {
            "keepPerRepo": int(OPTIONS.get("registryKeepPerRepo") or 20),
            "minAgeHours": float(OPTIONS.get("registryMinAgeHours") or 48),
            "protects": ["current workload refs", "digest closure", "protected tags", "recent tags", "newest N tags per repo"],
        },
    }
    return summary

def growth_watermark_policy(root_disk):
    use_percent = root_disk.get("usePercent") if isinstance(root_disk, dict) else None
    if use_percent is None:
        state = "unknown"
        action = "collect-snapshot"
    elif use_percent < 75:
        state = "healthy"
        action = "observe-trend"
    elif use_percent < 80:
        state = "watch"
        action = "run-dry-run-plan"
    elif use_percent < 85:
        state = "maintenance"
        action = "schedule-owner-aware-retention"
    else:
        state = "emergency"
        action = "restore-runtime-then-file-evidence"
    return {
        "state": state,
        "recommendedAction": action,
        "watermarks": [
            {"range": "<75%", "action": "trend only"},
            {"range": "75%-80%", "action": "run dry-run plan and identify source"},
            {"range": "80%-85%", "action": "small owner-aware retention run"},
            {"range": ">=85%", "action": "runtime recovery first, then root-cause growth source"},
        ],
        "growthThresholdPolicy": "If bytes/day remains high for consecutive snapshots, act before 80%; exact threshold should be set from the first week of saved snapshots.",
    }

def snapshot_metric_map(snapshot):
    metrics = {}
    root = snapshot.get("rootDisk") or {}
    if isinstance(root, dict) and root.get("usedBytes") is not None:
        metrics["root.usedBytes"] = {"value": safe_int(root.get("usedBytes")), "unit": "bytes", "label": "root used bytes"}
    for item in snapshot.get("sources") or []:
        if not isinstance(item, dict) or item.get("sizeBytes") is None:
            continue
        key = "source.%s.sizeBytes" % item.get("id")
        metrics[key] = {"value": safe_int(item.get("sizeBytes")), "unit": "bytes", "label": item.get("label") or item.get("id")}
    storage = ((snapshot.get("ciStorage") or {}).get("byOwnerGroup") or {})
    if not storage:
        storage = ((snapshot.get("pvcAttribution") or {}).get("byOwnerGroup") or {})
    for owner, value in storage.items():
        metrics["ciStorage.%s.estimatedBytes" % owner] = {"value": safe_int((value or {}).get("estimatedBytes")), "unit": "bytes", "label": "CI storage %s" % owner}
    memory = snapshot.get("memoryPressure") or {}
    memory_summary = memory.get("summary") or {}
    if memory_summary.get("matchedRssBytes") is not None:
        metrics["memoryPressure.matchedRssBytes"] = {"value": safe_int(memory_summary.get("matchedRssBytes")), "unit": "bytes", "label": "matched observer/chrome RSS"}
    if memory_summary.get("observeStateBytes") is not None:
        metrics["memoryPressure.observeStateBytes"] = {"value": safe_int(memory_summary.get("observeStateBytes")), "unit": "bytes", "label": "web observe state bytes"}
    for key in ["matchedProcessCount", "activeObserverSignals", "staleObserverSignals"]:
        if memory_summary.get(key) is not None:
            metrics["memoryPressure.%s" % key] = {"value": safe_int(memory_summary.get(key)), "unit": "count", "label": "memory pressure %s" % key}
    registry = snapshot.get("registry") or {}
    retention = registry.get("retentionPlan") or {}
    for key in ["totalTags", "totalRevisions", "deleteTags", "deleteRevisions", "estimatedReclaimBytes"]:
        if key in retention and retention.get(key) is not None:
            unit = "bytes" if key.endswith("Bytes") else "count"
            metrics["registry.%s" % key] = {"value": safe_int(retention.get(key)), "unit": unit, "label": "registry %s" % key}
    return metrics

def delta_metric_rows(before, after):
    before_metrics = snapshot_metric_map(before)
    after_metrics = snapshot_metric_map(after)
    before_ts = iso_to_epoch(before.get("observedAt"))
    after_ts = iso_to_epoch(after.get("observedAt"))
    seconds = (after_ts - before_ts) if before_ts is not None and after_ts is not None else None
    rows = []
    for key in sorted(set(before_metrics.keys()) | set(after_metrics.keys())):
        old = before_metrics.get(key, {"value": 0, "unit": (after_metrics.get(key) or {}).get("unit"), "label": key})
        new = after_metrics.get(key, {"value": 0, "unit": old.get("unit"), "label": old.get("label")})
        delta = safe_int(new.get("value")) - safe_int(old.get("value"))
        row = {
            "key": key,
            "label": new.get("label") or old.get("label") or key,
            "unit": new.get("unit") or old.get("unit"),
            "before": old.get("value"),
            "after": new.get("value"),
            "delta": delta,
        }
        if row["unit"] == "bytes":
            row["beforeHuman"] = fmt_bytes(row["before"] or 0)
            row["afterHuman"] = fmt_bytes(row["after"] or 0)
            row["deltaHuman"] = ("-" if delta < 0 else "") + fmt_bytes(abs(delta))
            if seconds and seconds > 0:
                per_day = int(delta * 86400 / seconds)
                row["perDayBytes"] = per_day
                row["perDayHuman"] = ("-" if per_day < 0 else "") + fmt_bytes(abs(per_day)) + "/day"
        rows.append(row)
    rows.sort(key=lambda item: safe_int(item.get("delta")), reverse=True)
    return {"durationSeconds": seconds, "metrics": rows}

def growth_trend_payload(points):
    points = [point for point in points if isinstance(point, dict)]
    if len(points) < 2:
        return {
            "pointCount": len(points),
            "state": "insufficient-history",
            "message": "Run snapshot at least twice to compute deltas.",
        }
    latest_delta = delta_metric_rows(points[-2], points[-1])
    window_delta = delta_metric_rows(points[0], points[-1])
    def rate_warning(delta):
        seconds = delta.get("durationSeconds")
        if seconds is not None and seconds < 3600:
            return {
                "code": "short-window-rate-noisy",
                "message": "Per-day rates from windows shorter than 1 hour are directional only; use daily snapshots for governance decisions.",
                "durationSeconds": seconds,
            }
        return None
    return {
        "pointCount": len(points),
        "oldestAt": points[0].get("observedAt"),
        "latestAt": points[-1].get("observedAt"),
        "latestDelta": {
            "durationSeconds": latest_delta.get("durationSeconds"),
            "rateWarning": rate_warning(latest_delta),
            "topGrowingBytes": [row for row in latest_delta.get("metrics", []) if row.get("unit") == "bytes" and safe_int(row.get("delta")) > 0][:10],
            "topShrinkingBytes": [row for row in reversed(latest_delta.get("metrics", [])) if row.get("unit") == "bytes" and safe_int(row.get("delta")) < 0][:10],
            "registryCounters": [row for row in latest_delta.get("metrics", []) if str(row.get("key", "")).startswith("registry.") and row.get("unit") == "count"],
        },
        "windowDelta": {
            "durationSeconds": window_delta.get("durationSeconds"),
            "rateWarning": rate_warning(window_delta),
            "topGrowingBytes": [row for row in window_delta.get("metrics", []) if row.get("unit") == "bytes" and safe_int(row.get("delta")) > 0][:10],
            "topShrinkingBytes": [row for row in reversed(window_delta.get("metrics", [])) if row.get("unit") == "bytes" and safe_int(row.get("delta")) < 0][:10],
            "registryCounters": [row for row in window_delta.get("metrics", []) if str(row.get("key", "")).startswith("registry.") and row.get("unit") == "count"],
        },
    }

def compact_metric_rows(rows, limit=3):
    compact = []
    for row in (rows or [])[:limit]:
        compact.append({
            "key": row.get("key"),
            "label": row.get("label"),
            "unit": row.get("unit"),
            "delta": row.get("delta"),
            "deltaHuman": row.get("deltaHuman"),
            "perDayHuman": row.get("perDayHuman"),
        })
    return compact

def compact_trend_payload(payload):
    if payload.get("state") == "insufficient-history":
        return payload
    latest = payload.get("latestDelta") or {}
    window = payload.get("windowDelta") or {}
    return {
        "pointCount": payload.get("pointCount"),
        "oldestAt": payload.get("oldestAt"),
        "latestAt": payload.get("latestAt"),
        "latestDelta": {
            "durationSeconds": latest.get("durationSeconds"),
            "rateWarning": latest.get("rateWarning"),
            "topGrowingBytes": compact_metric_rows(latest.get("topGrowingBytes") or [], 1),
            "topShrinkingBytes": compact_metric_rows(latest.get("topShrinkingBytes") or [], 1),
            "registryCounters": compact_metric_rows(latest.get("registryCounters") or [], 1),
        },
        "windowDelta": {
            "durationSeconds": window.get("durationSeconds"),
            "rateWarning": window.get("rateWarning"),
            "topGrowingBytes": compact_metric_rows(window.get("topGrowingBytes") or [], 1),
            "topShrinkingBytes": compact_metric_rows(window.get("topShrinkingBytes") or [], 1),
            "registryCounters": compact_metric_rows(window.get("registryCounters") or [], 1),
        },
        "fullDisclosure": "rerun trend --full for all metric rows",
    }

def compact_growth_point(item):
    registry = item.get("registry") or {}
    retention = registry.get("retentionPlan") or {}
    ci_storage = item.get("ciStorage") or {}
    containerd = item.get("containerd") or {}
    memory = item.get("memoryPressure") or {}
    memory_summary = memory.get("summary") or {}
    observe = (memory.get("webObserve") or {})
    return {
        "observedAt": item.get("observedAt"),
        "rootDisk": item.get("rootDisk"),
        "sourceCount": len(item.get("sources") or []),
        "registry": {
            "sizeBytes": registry.get("sizeBytes"),
            "sizeHuman": registry.get("sizeHuman"),
            "totalTags": retention.get("totalTags"),
            "totalRevisions": retention.get("totalRevisions"),
            "deleteTags": retention.get("deleteTags"),
            "deleteRevisions": retention.get("deleteRevisions"),
            "estimatedReclaimBytes": retention.get("estimatedReclaimBytes"),
            "estimatedReclaimHuman": retention.get("estimatedReclaimHuman"),
        },
        "ciStorage": {
            "pvcCount": ci_storage.get("pvcCount"),
            "estimatedBytes": ci_storage.get("estimatedBytes"),
            "estimatedHuman": ci_storage.get("estimatedHuman"),
            "byOwnerGroup": ci_storage.get("byOwnerGroup"),
        },
        "containerd": {
            "state": containerd.get("state"),
            "cleanupSupported": containerd.get("cleanupSupported"),
        },
        "memoryPressure": {
            "matchedProcessCount": memory_summary.get("matchedProcessCount"),
            "matchedRssBytes": memory_summary.get("matchedRssBytes"),
            "matchedRssHuman": memory_summary.get("matchedRssHuman"),
            "activeObserverSignals": memory_summary.get("activeObserverSignals"),
            "staleObserverSignals": memory_summary.get("staleObserverSignals"),
            "observeStateBytes": memory_summary.get("observeStateBytes"),
            "observeStateHuman": memory_summary.get("observeStateHuman"),
            "webObserveRootCount": observe.get("rootCount"),
        },
    }

def collect_growth_snapshot(observed_at, preflight):
    root_disk = df_snapshot()
    sources = disk_source_snapshot()
    ci_storage = ci_storage_snapshot()
    memory_pressure = collect_memory_pressure()
    compact_pvc = compact_pvc_attribution(ci_storage)
    if bool(OPTIONS.get("full")):
        public_pvc = ci_storage
        public_memory = memory_pressure
    else:
        public_pvc = compact_ci_storage_summary(ci_storage)
        public_memory = compact_memory_summary(memory_pressure)
    registry = registry_growth_snapshot()
    containerd = containerd_breakdown_snapshot()
    commands = {
        "snapshot": "bun scripts/cli.ts gc remote %s snapshot --include-hwlab-registry --history-limit %s" % (PROVIDER_ID, int(OPTIONS.get("historyLimit") or 12)),
        "trend": "bun scripts/cli.ts gc remote %s trend --history-limit %s" % (PROVIDER_ID, int(OPTIONS.get("historyLimit") or 12)),
        "registryPlan": "bun scripts/cli.ts gc remote %s plan --target-use-percent 70 --include-hwlab-registry --limit 50" % PROVIDER_ID,
        "hwlabCiRetention": ((ci_storage.get("handoff") or {}).get("hwlab") or {}).get("dryRun"),
        "agentrunRetention": ((ci_storage.get("handoff") or {}).get("agentrun") or {}).get("dryRun"),
        "remotePolicy": "bun scripts/cli.ts gc remote %s policy plan" % PROVIDER_ID,
    }
    if not bool(OPTIONS.get("full")):
        commands = {
            "trend": "bun scripts/cli.ts gc remote %s trend --history-limit %s" % (PROVIDER_ID, int(OPTIONS.get("historyLimit") or 12)),
            "status": "bun scripts/cli.ts gc remote %s status --limit %s" % (PROVIDER_ID, int(OPTIONS.get("limit") or 50)),
            "full": "bun scripts/cli.ts gc remote %s snapshot --full --no-save" % PROVIDER_ID,
        }
    return {
        "ok": True,
        "action": "gc remote snapshot",
        "providerId": PROVIDER_ID,
        "dryRun": True,
        "mutation": False,
        "diagnosticStateMutation": bool(OPTIONS.get("saveSnapshot", True)),
        "observedAt": observed_at,
        "rootDisk": root_disk,
        "clusterPreflight": preflight,
        "sources": sources,
        "registry": registry,
        "pvcAttribution": public_pvc,
        "memoryPressure": public_memory,
        "containerd": containerd,
        "policy": growth_watermark_policy(root_disk or {}),
        "commands": commands,
    }