"""CHV 上一世残留 VM 清理（生产路径唯一入口）。

本模块是叶子模块，只依赖标准库，绝不 import
cloud.chv.experimental_pool（实验接口）与 cloud.sandbox。
生产代码（worker 启动/关闭）只调
reconcile_orphans_for_executor()，它直接读 registry 文件、
对每个上一世条目调 executor.shutdown_vm()，不构造任何
池对象、不调 prewarm/acquire/release。

Registry 语义与旧 WarmPool._load_registry 完全一致
（feat-chv-arch95plus 的 fail-safe 口径，本轨只搬位置、
不改语义）：
- 文件：state_dir/warm-pool.json，内容为
  {<合成名>: {"state", "vm", "task_id", "snapshot_url"}}。
- 读到的 warm/assigned 条目一律当 orphan：旧 VMM 进程
  已随旧 worker 消失，无法验证它还是当初那个干净快照
  VM，绝不发给任务，只能清理。
- 损坏 fail-safe：文件不是合法 JSON / 顶层不是 dict 时，
  整个文件 quarantine 成 <path>.corrupt，清理从空开始；
  损坏本身发审计（feat-chv-p1fix，P1-02），绝不静默
  返回空——损坏文件的孤儿 VM 此前只能靠 socket 扫描。
  单条目结构非法（无 vm、或 state 不在 warm/assigned）
  时只丢该条，不连坐其它条目；非法条目发审计并计数
  （feat-chv-p1fix，P1-02），落盘时不再写回。
- 清理后落盘：全部成功时为空 dict；有失败项时只写回
  失败项的原始条目（原子 tmp + rename），下次 reconcile
  可重试，失败项绝不被“已清理”掩盖（feat-chv-p1fix，
  P1-01）。写失败只记日志警告，不炸 worker 启动/关闭。
  shutdown_vm 抛错的条目不进 cleaned、不触发“全清”，
  而是记 failed 列表并经 executor._emit 发审计；
  _emit 失败同样 best-effort。整个 reconcile 在
  state_dir 的 flock 下跑（见 locking.py），与 wake 互斥。
"""

import json
import logging
import os

_LOGGER = logging.getLogger("cloud.cloud_hypervisor")

_REGISTRY_FILENAME = "warm-pool.json"


def registry_path_for_executor(executor: object) -> str | None:
    """执行器的 registry 路径：state_dir 下的 warm-pool.json。

    无 state_dir 时返回 None（调用方按“无残留”处理）。"""
    state_dir = getattr(executor, "state_dir", None)
    if not state_dir:
        return None
    return os.path.join(state_dir, _REGISTRY_FILENAME)


def _quarantine_registry(path: str) -> None:
    """把损坏的 registry 文件挪到 <path>.corrupt 旁（尽力而为）。"""
    try:
        os.replace(path, path + ".corrupt")
    except OSError:
        _LOGGER.warning("chv_pool_registry_quarantine_failed "
                        "path=%s", path)


def _read_registry(path: str) -> tuple[dict | None, bool]:
    """读 registry 原始 dict：(data, corrupt)。

    无文件 → (None, False)；损坏 → quarantine 后 (None, True)；
    顶层非 dict 同样算损坏。调用方按 corrupt 决定是否记
    registry_corrupt（实验池需要区分“无文件”与“损坏”）。"""
    try:
        with open(path, "r") as fh:
            data = json.load(fh)
    except FileNotFoundError:
        return None, False
    except (OSError, ValueError):
        _quarantine_registry(path)
        return None, True
    if not isinstance(data, dict):
        _quarantine_registry(path)
        return None, True
    return data, False


def _write_registry(path: str, data: dict) -> None:
    """原子落盘 registry（tmp + rename）；写失败只记日志。"""
    try:
        directory = os.path.dirname(path)
        if directory:
            os.makedirs(directory, exist_ok=True)
        tmp_path = path + ".tmp"
        with open(tmp_path, "w") as fh:
            json.dump(data, fh)
        os.replace(tmp_path, path)
    except OSError:
        _LOGGER.warning("chv_pool_registry_save_failed path=%s", path)


def _orphan_entries(data: dict) -> list[dict]:
    """从原始 registry dict 里筛出可清理的上一世条目。

    只留 state 为 warm/assigned 且有 vm 的条目；非法条目
    不连坐其它条目（计数与审计见 reconcile 本体，不再
    静默丢弃）。返回条目为 {"vm", "task_id",
    "snapshot_url"} 的新 dict，不改原数据。"""
    entries: list[dict] = []
    for _key, entry in _orphan_entries_with_keys(data)[0]:
        entries.append(entry)
    return entries


def _orphan_entries_with_keys(data: dict) -> tuple[list, list]:
    """筛条目并保留原始 registry 键：返回（可清理条目，
    非法条目键）。

    可清理条目为 (原始键, {"vm", "task_id",
    "snapshot_url"})；非法条目只记键（键本身不含密钥，
    可进审计 detail），原始值不进审计。"""
    entries: list = []
    invalid_keys: list = []
    for key, entry in data.items():
        if not isinstance(entry, dict):
            invalid_keys.append(key)
            continue
        if not entry.get("vm"):
            invalid_keys.append(key)
            continue
        if entry.get("state") not in ("warm", "assigned"):
            invalid_keys.append(key)
            continue
        entries.append((key, {
            "vm": entry["vm"],
            "task_id": entry.get("task_id"),
            "snapshot_url": entry.get("snapshot_url"),
        }))
    return entries, invalid_keys


def _best_effort_emit(executor: object, event: str, detail: str) -> None:
    """经 executor._emit 发审计（best-effort，兼容只接受
    (event, task, detail) 三参的假执行器；无 _emit 时跳过）。"""
    emit = getattr(executor, "_emit", None)
    if not callable(emit):
        return
    try:
        emit(event, {"id": None}, detail)
    except Exception:
        pass


def reconcile_orphans_for_executor(executor: object) -> list[str]:
    """生产路径唯一的残留清理入口：返回已清理的 VM 名列表。

    只做清理：读 registry → 每个上一世条目尽力
    shutdown_vm → 落盘 → best-effort 审计事件。
    绝不构造 WarmPool、绝不调 prewarm/acquire/release，
    绝不把任何条目发给任务。无 registry 路径时返回 []。

    feat-chv-p1fix（P1-01/P1-02/P1-05）语义：
    - shutdown 失败的条目不进 cleaned，记 failed 并审计，
      落盘时只写回失败项原始条目（可重试），绝不清库；
    - registry 损坏时 quarantine 后仍发审计，不静默返回空；
    - 非法条目发审计并计数，落盘时丢弃（已审计，不再
      每次重启重复审计同一批非法条目）；
    - 全程持 state_dir flock，与 wake/restore/cleanup 互斥。
      返回值仍只含 cleaned（调用方兼容不变）；失败与
      损坏/非法的可见性走 executor._emit（worker 接线后
      即任务 timeline 审计），不靠返回值传递。
    """
    path = registry_path_for_executor(executor)
    if not path:
        return []
    if not os.path.exists(path):
        return []
    state_dir = getattr(executor, "state_dir", None)
    if state_dir:
        from cloud.chv.locking import StateLock
        lock = StateLock(state_dir)
    else:  # pragma: no cover - state_dir 缺失时 path 已为 None
        from contextlib import nullcontext
        lock = nullcontext()  # type: ignore[assignment]
    with lock:
        return _reconcile_locked(executor, path)


def _reconcile_locked(executor: object, path: str) -> list[str]:
    """持锁的 reconcile 本体（见 reconcile_orphans_for_executor）。"""
    data, corrupt = _read_registry(path)
    if data is None:
        if corrupt:
            _best_effort_emit(
                executor, "warm_pool_registry_corrupt",
                "path=%s registry corrupt, quarantined" % path)
        return []
    entries, invalid_keys = _orphan_entries_with_keys(data)
    if invalid_keys:
        _best_effort_emit(
            executor, "warm_pool_registry_invalid_entries",
            "invalid=%d keys=%s"
            % (len(invalid_keys), ",".join(str(k) for k in invalid_keys)))
    cleaned: list[str] = []
    failed: list[str] = []
    failed_errors: list[str] = []
    failed_keys: list[str] = []
    for key, entry in entries:
        vm = entry["vm"]
        try:
            executor.shutdown_vm(vm)  # type: ignore[attr-defined]
        except Exception as exc:
            failed.append(vm)
            failed_keys.append(key)
            failed_errors.append("%s:%s" % (vm, type(exc).__name__))
            continue
        cleaned.append(vm)
    if cleaned or failed or invalid_keys:
        remaining = {key: data[key] for key in failed_keys
                     if key in data}
        _write_registry(path, remaining)
    if cleaned:
        _best_effort_emit(
            executor, "warm_pool_reconcile",
            "cleaned=%s" % ",".join(cleaned))
    if failed:
        _best_effort_emit(
            executor, "warm_pool_reconcile_failed",
            "failed=%s cleaned=%s errors=%s"
            % (",".join(failed), ",".join(cleaned),
               ";".join(failed_errors)))
    return cleaned