from __future__ import annotations

import json
import logging
import os
import uuid
from datetime import datetime, timedelta, timezone
from pathlib import Path
from typing import Any

from niannian.core.schemas.models import Atom
from niannian.storage.archive_bundle import (
    build_clean_bundle,
    bundle_contains_forbidden_fields,
    select_archive_batch,
    select_distilled_candidates,
)
from niannian.storage.baidu_backup import read_backup_state
from niannian.storage.cloud_dual import CloudDualBackend
from niannian.storage.cloud_factory import build_cloud_backend
from niannian.storage.cloud_mirror import read_mirror_state
from niannian.storage.cloud_tier import CloudTierRouter
from niannian.storage.manifest import Manifest, ManifestConflictError, ManifestManager, sha256_file, utc_now_iso
from niannian.storage.merge import merge_atoms_jsonl, merge_json_dict
from niannian.storage.ops_error_log import OpsErrorLog
from niannian.storage.primary_backup import PrimaryBackupBackend
from niannian.storage.sync.conditions import evaluate_push_conditions, parse_iso
from niannian.storage.types import ArchiveResult, FlushResult, PullResult, SyncStatus

logger = logging.getLogger(__name__)
from niannian.storage.webdav import webdav_put


class LocalMemoryStore:
    """L0 热缓存 — 本地 JSONL/JSON 读写 + L1 双写。"""

    TRACKED_FILES = (
        "left/atoms.jsonl",
        "cerebellum/pages.json",
        "cerebellum/graph.json",
        "cerebellum/compile_state.json",
        "hippocampus/summaries.json",
        "hippocampus/lifecycle.json",
        "sync/ops_errors.jsonl",
    )
    VAULT_STATIC_FILES = (
        "vault/_index.json",
        "vault/_rules/routing_rules.yaml",
        "vault/_embeddings.jsonl",
        "vault/_fts.sqlite",
    )

    def __init__(
        self,
        home: Path,
        *,
        node_id: str | None = None,
        config: dict[str, Any] | None = None,
        providers: dict[str, Any] | None = None,
    ) -> None:
        self._home = home
        self._config = config or {}
        self._providers = providers or {}
        self._sync_cfg = self._config.get("sync") or {}
        self._node_id = node_id or self._default_node_id()
        self._atoms_path = home / "left" / "atoms.jsonl"
        self._pages_path = home / "cerebellum" / "pages.json"
        self._graph_path = home / "cerebellum" / "graph.json"
        self._compile_state_path = home / "cerebellum" / "compile_state.json"
        self._summaries_path = home / "hippocampus" / "summaries.json"
        self._lifecycle_path = home / "hippocampus" / "lifecycle.json"
        self._pending_path = home / "sync" / "pending.jsonl"
        self._state_path = home / "sync" / "state.json"
        self._manifest = ManifestManager(home / "sync" / "manifest.json")
        self._cloud = build_cloud_backend(home, config=self._config, providers=self._providers)
        self._ensure_dirs()

    @property
    def cloud(self) -> CloudDualBackend | PrimaryBackupBackend | CloudTierRouter:
        return self._cloud

    def _default_node_id(self) -> str:
        mode = os.environ.get("NIANNIAN_MODE") or (self._config.get("runtime") or {}).get("mode", "auto")
        if mode == "docker":
            return "docker-nas"
        if mode == "mac":
            return f"mac-{uuid.uuid4().hex[:8]}"
        return f"local-{uuid.uuid4().hex[:8]}"

    def _runtime_node_key(self) -> str:
        mode = os.environ.get("NIANNIAN_MODE") or (self._config.get("runtime") or {}).get("mode", "auto")
        return "docker" if mode == "docker" else "mac"

    @property
    def home(self) -> Path:
        return self._home

    @property
    def manifest(self) -> ManifestManager:
        return self._manifest

    def _ensure_dirs(self) -> None:
        for d in ("left", "cerebellum", "hippocampus", "sync"):
            (self._home / d).mkdir(parents=True, exist_ok=True)
        if not self._atoms_path.exists():
            self._atoms_path.touch()
        if not self._pages_path.exists():
            self._pages_path.write_text('{"version": 2, "pages": {}}\n', encoding="utf-8")
        if not self._graph_path.exists():
            self._graph_path.write_text('{"version": 1, "nodes": {}, "edges": []}\n', encoding="utf-8")
        if not self._compile_state_path.exists():
            self._compile_state_path.write_text("{}\n", encoding="utf-8")
        if not self._summaries_path.exists():
            self._summaries_path.write_text("{}\n", encoding="utf-8")
        if not self._lifecycle_path.exists():
            self._lifecycle_path.write_text("{}\n", encoding="utf-8")
        if not self._pending_path.exists():
            self._pending_path.touch()
        if not self._state_path.exists():
            self._write_state({"node_id": self._node_id, "last_sync": None})

    def _path_for_rel(self, rel_path: str) -> Path:
        return self._home / rel_path

    def _read_json(self, path: Path) -> dict:
        if not path.is_file():
            return {}
        return json.loads(path.read_text(encoding="utf-8"))

    def _write_json(self, path: Path, data: dict) -> None:
        path.parent.mkdir(parents=True, exist_ok=True)
        path.write_text(
            json.dumps(data, ensure_ascii=False, indent=2) + "\n",
            encoding="utf-8",
        )

    def _read_state(self) -> dict:
        return self._read_json(self._state_path)

    def _write_state(self, data: dict) -> None:
        self._write_json(self._state_path, data)

    def _atom_from_dict(self, d: dict) -> Atom:
        return Atom(
            id=d["id"],
            raw_text=d["raw_text"],
            tags=d.get("tags") or [],
            source=d.get("source", "manual"),
            project=d.get("project", "global"),
            created_at=d["created_at"],
            status=d.get("status", "raw"),
            meta=d.get("meta") or {},
        )

    def read_atoms(
        self,
        project: str | None = None,
        *,
        limit: int = 500,
        status: str | None = None,
    ) -> list[Atom]:
        if not self._atoms_path.is_file():
            return []
        rows: list[Atom] = []
        with self._atoms_path.open(encoding="utf-8") as f:
            for line in f:
                line = line.strip()
                if not line:
                    continue
                d = json.loads(line)
                if project and d.get("project") != project:
                    continue
                if status and d.get("status") != status:
                    continue
                rows.append(self._atom_from_dict(d))
        return rows[-limit:]

    def append_atom(self, atom: Atom) -> None:
        with self._atoms_path.open("a", encoding="utf-8") as f:
            f.write(json.dumps(atom.to_dict(), ensure_ascii=False) + "\n")
        pending = {
            "atom_id": atom.id,
            "path": "left/atoms.jsonl",
            "ts": utc_now_iso(),
        }
        with self._pending_path.open("a", encoding="utf-8") as f:
            f.write(json.dumps(pending, ensure_ascii=False) + "\n")
        self._manifest.update_file_entry("left/atoms.jsonl", self._atoms_path)

    def patch_atom_meta(self, atom_id: str, meta_patch: dict[str, Any]) -> Atom | None:
        if not self._atoms_path.is_file():
            return None
        from niannian.memory.belief import normalize_belief_status
        from niannian.memory.meta import parse_atom_meta

        if "belief_status" in (meta_patch or {}):
            normalize_belief_status(meta_patch["belief_status"])

        updated: Atom | None = None
        out_lines: list[str] = []
        with self._atoms_path.open(encoding="utf-8") as f:
            for line in f:
                raw = line.strip()
                if not raw:
                    continue
                row = json.loads(raw)
                if row.get("id") != atom_id:
                    out_lines.append(raw)
                    continue
                meta = dict(row.get("meta") or {})
                patch = dict(meta_patch or {})
                if "provenance" in patch and isinstance(patch["provenance"], dict):
                    prov = dict(meta.get("provenance") or {})
                    prov.update(patch["provenance"])
                    meta["provenance"] = prov
                    patch = {k: v for k, v in patch.items() if k != "provenance"}
                meta.update(patch)
                parsed = parse_atom_meta(meta)
                row_meta = parsed.to_dict()
                for key, val in meta.items():
                    if key not in row_meta:
                        row_meta[key] = val
                row["meta"] = row_meta
                updated = self._atom_from_dict(row)
                out_lines.append(json.dumps(row, ensure_ascii=False))
        if updated is None:
            return None
        self._atoms_path.write_text("\n".join(out_lines) + "\n", encoding="utf-8")
        self._manifest.update_file_entry("left/atoms.jsonl", self._atoms_path)
        return updated

    def read_wiki_pages_raw(self) -> dict:
        return self._read_json(self._pages_path)

    def write_wiki_pages_document(self, doc: dict) -> None:
        self._write_json(self._pages_path, doc)
        self._manifest.update_file_entry("cerebellum/pages.json", self._pages_path)

    def read_wiki_graph_raw(self) -> dict:
        return self._read_json(self._graph_path)

    def write_wiki_graph(self, doc: dict) -> None:
        self._write_json(self._graph_path, doc)
        self._manifest.update_file_entry("cerebellum/graph.json", self._graph_path)

    def read_wiki_compile_state(self) -> dict:
        return self._read_json(self._compile_state_path)

    def write_wiki_compile_state(self, doc: dict) -> None:
        self._write_json(self._compile_state_path, doc)
        self._manifest.update_file_entry("cerebellum/compile_state.json", self._compile_state_path)

    def read_wiki_pages(self) -> dict:
        from niannian.brain.cerebellum.wiki_store import WikiStore

        return WikiStore(self).pages_map()

    def update_wiki_pages(self, patch: dict) -> None:
        from niannian.brain.cerebellum.wiki_store import WikiStore

        WikiStore(self).patch_pages(patch)
        self._manifest.update_file_entry("cerebellum/pages.json", self._pages_path)

    def read_summaries(self) -> dict:
        return self._read_json(self._summaries_path)

    def write_summaries(self, data: dict) -> None:
        self._write_json(self._summaries_path, data)
        self._manifest.update_file_entry("hippocampus/summaries.json", self._summaries_path)

    def read_lifecycle(self) -> dict:
        return self._read_json(self._lifecycle_path)

    def write_lifecycle(self, data: dict) -> None:
        self._write_json(self._lifecycle_path, data)
        self._manifest.update_file_entry("hippocampus/lifecycle.json", self._lifecycle_path)

    def _count_pending(self) -> tuple[int, int]:
        if not self._pending_path.is_file():
            return 0, 0
        count = 0
        size = self._pending_path.stat().st_size
        with self._pending_path.open(encoding="utf-8") as f:
            for line in f:
                if line.strip():
                    count += 1
        return count, size

    def _clear_pending(self) -> None:
        self._pending_path.write_text("", encoding="utf-8")

    def _vault_md_rels(self) -> list[str]:
        root = self._home / "vault"
        if not root.is_dir():
            return []
        return [
            str(p.relative_to(self._home).as_posix())
            for p in sorted(root.rglob("*.md"))
            if p.is_file()
        ]

    def _sync_file_rels(self) -> list[str]:
        seen: set[str] = set()
        rels: list[str] = []
        for rel in (*self.TRACKED_FILES, *self.VAULT_STATIC_FILES, *self._vault_md_rels()):
            if rel not in seen:
                seen.add(rel)
                rels.append(rel)
        return rels

    def _refresh_local_manifest_entries(self) -> Manifest:
        from niannian.storage.manifest import FileEntry

        manifest = self._manifest.read()
        for rel in self._sync_file_rels():
            path = self._path_for_rel(rel)
            if path.is_file():
                manifest.files[rel] = FileEntry(
                    sha256=sha256_file(path),
                    size=path.stat().st_size,
                    tier="hot",
                    updated_at=utc_now_iso(),
                )
        return manifest

    def _apply_remote_file(self, rel_path: str, data: bytes) -> None:
        local_path = self._path_for_rel(rel_path)
        if rel_path == "left/atoms.jsonl":
            merge_atoms_jsonl(local_path, data)
        elif rel_path.endswith(".json"):
            merge_json_dict(local_path, data)
        else:
            local_path.parent.mkdir(parents=True, exist_ok=True)
            local_path.write_bytes(data)
        self._manifest.update_file_entry(rel_path, local_path)

    def _vault_index_entry_count(self) -> int:
        from niannian.vault.enhancements import vault_entry_count

        idx_path = self._home / "vault" / "_index.json"
        if not idx_path.is_file():
            return 0
        try:
            data = json.loads(idx_path.read_text(encoding="utf-8"))
        except (json.JSONDecodeError, OSError):
            return 0
        return vault_entry_count(data if isinstance(data, dict) else {})

    def _incremental_vault_sync_active(self) -> bool:
        from niannian.vault.enhancements import FEATURE_INCREMENTAL_SYNC, enhancement_active

        return enhancement_active(
            self._config,
            FEATURE_INCREMENTAL_SYNC,
            entry_count=self._vault_index_entry_count(),
        )

    def _vault_rel_unchanged_on_remote(self, rel: str, local_manifest: Manifest, remote: Manifest | None) -> bool:
        if not rel.startswith("vault/") or not rel.endswith(".md"):
            return False
        if remote is None:
            return False
        local_entry = local_manifest.files.get(rel)
        remote_entry = remote.files.get(rel)
        if local_entry is None or remote_entry is None:
            return False
        return local_entry.sha256 == remote_entry.sha256

    def evaluate_push(self, *, force: bool = False) -> tuple[bool, list[str]]:
        pending_atoms, pending_bytes = self._count_pending()
        state = self._read_state()
        return evaluate_push_conditions(
            sync_cfg=self._sync_cfg,
            pending_atoms=pending_atoms,
            pending_bytes=pending_bytes,
            last_sync=state.get("last_sync"),
            force=force,
        )

    def sync_status(self, *, probe_tier: bool = False) -> SyncStatus:
        state = self._read_state()
        manifest = self._manifest.read()
        pending_atoms, pending_bytes = self._count_pending()
        mode = os.environ.get("NIANNIAN_MODE", "auto")
        push_allowed, blocked = self.evaluate_push(force=False)
        last_sync = state.get("last_sync")
        manifest_versions = (
            self._cloud.manifest_versions()
            if probe_tier or not isinstance(self._cloud, CloudTierRouter)
            else self._cloud.fast_manifest_versions()
        )
        backends = self._cloud.backend_status()
        backends["mode"] = mode
        primary_mode = self._cloud.primary_mode()
        backends["primary_mode"] = primary_mode
        meta_path = self._home / "meta.json"
        last_dream_at = None
        if meta_path.is_file():
            try:
                last_dream_at = json.loads(meta_path.read_text(encoding="utf-8")).get("last_dream_at")
            except (json.JSONDecodeError, OSError):
                last_dream_at = None
        backup_state = read_backup_state(self._home)
        backup_info: dict[str, Any] = {}
        degraded = self._cloud.last_upload_degraded
        tier_info: dict[str, Any] = {}
        if isinstance(self._cloud, CloudTierRouter):
            tier_info = self._cloud.tier_summary(read_mirror_state(self._home), live_probe=probe_tier)
            if probe_tier:
                active = tier_info.get("active_role", "quark")
                degraded = degraded or active not in ("quark",)
        elif isinstance(self._cloud, PrimaryBackupBackend):
            backup_info = self._cloud.backup_status(backup_state)
        archive_info = (
            self._archive_status(state)
            if probe_tier
            else {
                "pending_distilled": state.get("archive_pending_distilled"),
                "last_skipped_reason": state.get("last_archive_skipped_reason"),
                "min_distilled_atoms": int(
                    ((self._config.get("storage") or {}).get("archive") or {}).get("min_distilled_atoms", 50)
                ),
            }
        )
        if probe_tier and "min_distilled_atoms" not in archive_info:
            archive_info["min_distilled_atoms"] = int(
                ((self._config.get("storage") or {}).get("archive") or {}).get("min_distilled_atoms", 50)
            )
        elif not probe_tier and archive_info.get("min_distilled_atoms") is None:
            archive_info["min_distilled_atoms"] = int(
                ((self._config.get("storage") or {}).get("archive") or {}).get("min_distilled_atoms", 50)
            )
        return SyncStatus(
            version=manifest.version,
            pending_atoms=pending_atoms,
            pending_bytes=pending_bytes,
            last_sync=last_sync,
            last_cloud_sync=last_sync,
            last_archive_at=state.get("last_archive_at"),
            last_dream_at=last_dream_at,
            realtime="quark",
            backup=backup_info,
            tier=tier_info,
            archive=archive_info,
            node_id=state.get("node_id", self._node_id),
            cloud_ready=self._cloud.cloud_ready(),
            push_allowed=push_allowed,
            push_blocked_reasons=blocked,
            backends=backends,
            manifest_versions=manifest_versions,
            primary_mode=primary_mode,
            degraded=degraded,
        )

    def _archive_status(self, state: dict[str, Any]) -> dict[str, Any]:
        archive_cfg = (self._config.get("storage") or {}).get("archive") or {}
        lifecycle = self.read_lifecycle()
        atoms = self.read_atoms(limit=10_000)
        tiers = set(archive_cfg.get("tiers") or ["distilled"])
        pending = 0
        for atom in atoms:
            lc = lifecycle.get(atom.id, {})
            status = lc.get("status", atom.status)
            if status not in tiers:
                continue
            if lc.get("nas_archived"):
                continue
            if archive_cfg.get("require_dream", True) and not lc.get("dream_at"):
                continue
            pending += 1
        return {
            "pending_distilled": pending,
            "last_skipped_reason": state.get("last_archive_skipped_reason"),
        }

    def flush_to_cloud(self, *, force: bool = False) -> FlushResult:
        if isinstance(self._cloud, CloudTierRouter):
            self._cloud.check_health(record_failure=True)
        if not self._cloud.cloud_ready():
            return FlushResult(
                ok=False,
                errors=["cloud backend not configured — set quark token or mirror root"],
            )

        allowed, blocked = self.evaluate_push(force=force)
        if not allowed:
            return FlushResult(
                ok=False,
                skipped=True,
                blocked_reasons=blocked,
                errors=[f"push blocked: {', '.join(blocked)}"],
            )

        errors: list[str] = []
        uploaded: list[str] = []

        remote = self._cloud.read_remote_manifest()
        local = self._manifest.read()
        if remote and remote.version > local.version:
            pull = self.pull_from_cloud()
            if not pull.ok:
                errors.extend(pull.errors)
            local = self._manifest.read()

        manifest = self._refresh_local_manifest_entries()
        node_key = self._runtime_node_key()
        manifest.nodes[node_key] = {
            "last_sync": utc_now_iso(),
            "node_id": self._node_id,
        }

        incremental_vault = self._incremental_vault_sync_active()

        for rel in self._sync_file_rels():
            path = self._path_for_rel(rel)
            if not path.is_file():
                continue
            if incremental_vault and self._vault_rel_unchanged_on_remote(rel, manifest, remote):
                continue
            result = self._cloud.dual_upload(rel, path.read_bytes())
            if not result.ok:
                errors.append(f"dual upload failed for {rel}")
            else:
                uploaded.append(rel)

        manifest.updated_at = utc_now_iso()
        manifest.version = max(local.version, remote.version if remote else 0) + 1
        manifest_result = self._cloud.upload_manifest(manifest)
        if not manifest_result.ok:
            errors.append("manifest dual upload failed")

        try:
            self._manifest.write(manifest, expected_version=local.version)
        except ManifestConflictError as exc:
            errors.append(str(exc))

        versions = self._cloud.manifest_versions()
        if isinstance(self._cloud, CloudDualBackend):
            q_v, b_v = versions.get("quark"), versions.get("baidu")
            if q_v is not None and b_v is not None and q_v != b_v:
                errors.append(f"manifest version mismatch: quark={q_v} baidu={b_v}")

        now = utc_now_iso()
        state = self._read_state()
        state["last_sync"] = now
        state["node_id"] = self._node_id
        self._write_state(state)

        ok = not errors and bool(uploaded)
        if ok:
            self._clear_pending()
        elif errors:
            OpsErrorLog(self._home, config=self._config).record(
                "cloud_push_failed",
                component="sync",
                message="; ".join(errors[:3]),
                detail={"errors": errors[:10], "uploaded": uploaded},
                check_key="cloud_push_failed",
            )

        return FlushResult(
            ok=ok,
            version=self._manifest.read().version,
            files_uploaded=uploaded,
            errors=errors,
            blocked_reasons=blocked if not ok and not uploaded else [],
        )

    def pull_from_cloud(self) -> PullResult:
        if not self._cloud.cloud_ready():
            return PullResult(
                ok=False,
                errors=["cloud backends not configured"],
            )

        remote = self._cloud.read_remote_manifest()
        if remote is None:
            return PullResult(ok=True, version=self._manifest.read().version, errors=["no remote manifest"])

        errors: list[str] = []
        downloaded: list[str] = []
        local = self._manifest.read()

        for rel, entry in remote.files.items():
            local_path = self._path_for_rel(rel)
            local_hash = sha256_file(local_path) if local_path.is_file() else None
            if local_hash == entry.sha256:
                continue
            data = self._cloud.dual_download(rel)
            if data is None:
                errors.append(f"download failed: {rel}")
                continue
            self._apply_remote_file(rel, data)
            downloaded.append(rel)

        if remote.version >= local.version:
            try:
                self._manifest.write(remote, expected_version=local.version)
            except ManifestConflictError:
                merged = self._manifest.read()
                if remote.version > merged.version:
                    self._manifest.write(remote)

        if errors:
            OpsErrorLog(self._home, config=self._config).record(
                "cloud_pull_failed",
                component="sync",
                message="; ".join(errors[:3]),
                detail={"errors": errors[:10], "downloaded": downloaded},
                check_key="cloud_pull_failed",
            )

        return PullResult(
            ok=not errors,
            version=self._manifest.read().version,
            files_downloaded=downloaded,
            errors=errors,
        )

    def archive_distilled(self, *, min_age_days: int = 7, archive_dir: Path | None = None) -> ArchiveResult:
        archive_cfg = (self._config.get("storage") or {}).get("archive") or {}
        nas_cfg = (self._sync_cfg.get("push_to_nas") or {})
        cfg = {
            "min_age_days": int(min_age_days or archive_cfg.get("min_age_days", 7)),
            "min_distilled_atoms": int(archive_cfg.get("min_distilled_atoms", 50)),
            "defer_extra_days": int(archive_cfg.get("defer_extra_days", 7)),
            "require_dream": bool(archive_cfg.get("require_dream", True)),
        }
        archive_root = archive_dir or Path(os.environ.get("NIANNIAN_ARCHIVE_DIR", "/data/archive"))
        min_distilled = cfg["min_distilled_atoms"]

        if nas_cfg.get("require_cloud_synced", True):
            state = self._read_state()
            if not state.get("last_sync"):
                return ArchiveResult(ok=False, errors=["require_cloud_synced: no cloud sync yet"])

        if nas_cfg.get("require_dream_completed", True) and cfg["require_dream"]:
            meta_path = self._home / "meta.json"
            last_dream = None
            if meta_path.is_file():
                try:
                    last_dream = json.loads(meta_path.read_text(encoding="utf-8")).get("last_dream_at")
                except (json.JSONDecodeError, OSError):
                    last_dream = None
            if not last_dream:
                return ArchiveResult(
                    ok=True,
                    chunks_archived=0,
                    skipped_reason="dream_not_completed",
                    min_distilled_atoms=min_distilled,
                )

        try:
            archive_root.mkdir(parents=True, exist_ok=True)
        except OSError as exc:
            return ArchiveResult(ok=False, errors=[f"archive dir not writable: {archive_root}: {exc}"])

        lifecycle = self.read_lifecycle()
        atoms = self.read_atoms(limit=10_000)
        now = datetime.now(timezone.utc)
        tiers = set(archive_cfg.get("tiers") or ["distilled"])

        candidates = select_distilled_candidates(
            atoms,
            lifecycle,
            require_dream=cfg["require_dream"],
            tiers=tiers,
        )
        to_archive, batch_meta = select_archive_batch(candidates, lifecycle, cfg, now)

        if not to_archive:
            skipped = batch_meta.get("skipped_reason")
            if skipped:
                state = self._read_state()
                state["last_archive_skipped_reason"] = skipped
                self._write_state(state)
            return ArchiveResult(
                ok=True,
                chunks_archived=0,
                bytes_archived=0,
                skipped_reason=skipped,
                eligible_at_7d=int(batch_meta.get("eligible_at_7d", 0)),
                min_distilled_atoms=min_distilled,
                next_retry_after_days=batch_meta.get("next_retry_after_days"),
            )

        summaries = self.read_summaries()
        wiki_pages = self.read_wiki_pages()
        project = self._config.get("project", {}).get("default_scope", "global")
        bundle = build_clean_bundle(
            atoms=to_archive,
            lifecycle=lifecycle,
            summaries=summaries,
            wiki_pages=wiki_pages,
            project=project,
            now=now,
        )
        if bundle_contains_forbidden_fields(bundle):
            return ArchiveResult(ok=False, errors=["clean bundle contains forbidden raw fields"])

        month_dir = archive_root / now.strftime("%Y-%m")
        month_dir.mkdir(parents=True, exist_ok=True)
        bundle_name = f"clean-{now.strftime('%Y%m%dT%H%M%S')}.json"
        bundle_path = month_dir / bundle_name
        payload = json.dumps(bundle, ensure_ascii=False, indent=2) + "\n"
        bundle_path.write_text(payload, encoding="utf-8")
        bytes_archived = bundle_path.stat().st_size

        payload_bytes = payload.encode("utf-8")
        webdav_ok: bool | None = None
        webdav_cfg = archive_cfg.get("webdav") or {}
        if webdav_cfg.get("enabled"):
            import os

            from niannian.config import resolve_env_value

            url = webdav_cfg.get("url")
            password = resolve_env_value(webdav_cfg.get("password"))
            remote_path = str(webdav_cfg.get("remote_path") or "/niannian/archive").strip("/")
            month_remote = f"{remote_path}/{now.strftime('%Y-%m')}" if remote_path else now.strftime("%Y-%m")
            try:
                webdav_ok = webdav_put(
                    base_url=str(url),
                    remote_path=month_remote,
                    rel_name=bundle_name,
                    data=payload_bytes,
                    username=webdav_cfg.get("username"),
                    password=password,
                )
            except Exception as exc:
                webdav_ok = False
                errors_webdav = str(exc)
            else:
                errors_webdav = None
        else:
            errors_webdav = None

        stamp = utc_now_iso()
        for atom in to_archive:
            lifecycle[atom.id] = {
                **lifecycle.get(atom.id, {}),
                "status": "distilled",
                "dream_at": lifecycle.get(atom.id, {}).get("dream_at"),
                "updated_at": stamp,
                "nas_archived": True,
            }
        self.write_lifecycle(lifecycle)

        state = self._read_state()
        state["nas_archived"] = True
        state["last_archive_at"] = stamp
        state.pop("last_archive_skipped_reason", None)
        self._write_state(state)

        manifest = self._manifest.read()
        manifest.nodes.setdefault("archive", {})["nas_archived"] = "true"
        manifest.nodes["archive"]["last_archive_at"] = stamp
        manifest.nodes["archive"]["bundle_format"] = "clean-v1"
        self._manifest.write(manifest)

        result_errors: list[str] = []
        if webdav_ok is False and errors_webdav:
            result_errors.append(f"webdav upload failed: {errors_webdav}")

        return ArchiveResult(
            ok=True,
            chunks_archived=1,
            bytes_archived=bytes_archived,
            eligible_at_7d=int(batch_meta.get("eligible_at_7d", 0)),
            min_distilled_atoms=min_distilled,
            webdav_ok=webdav_ok,
            errors=result_errors,
        )
