"""Cache-state discipline for measured rows.

Cache state is a measurement, not a wish (ADR-0002).  A row may claim "cold"

only when an OS page-cache purge verifiably succeeded, and "warm" only when

the caller explicitly performed the warming.  Everything else is reported as

"unknown" with a reason — never a silently fabricated state.

"""

from __future__ import annotations

import os

import platform

import subprocess

from pathlib import Path

from typing import Any

PURGE_TIMEOUT_S = 30

DARWIN_PURGE = "/usr/sbin/purge"

LINUX_DROP_CACHES = Path("/proc/sys/vm/drop_caches")

def _purge_result(attempted: bool, ok: bool, method: str | None, error: str | None) -> dict[str, Any]:

    return {"attempted": attempted, "ok": ok, "method": method, "error": error}

def _run_purge_command(command: list[str]) -> str | None:

    """Run a purge command; return None on success, an error string on failure."""

    try:

        completed = subprocess.run(command, capture_output=True, text=True, timeout=PURGE_TIMEOUT_S)

    except Exception as error:

        return f"{type(error).__name__}: {error}"

    if completed.returncode != 0:

        detail = (completed.stderr or completed.stdout or "").strip()

        return f"exit {completed.returncode}" + (f": {detail}" if detail else "")

    return None

def _drop_linux_caches(path: Path = LINUX_DROP_CACHES) -> str | None:

    """Write to /proc/sys/vm/drop_caches; return None on success, error string on failure."""

    try:

        os.sync()

        path.write_text("3\n")

    except Exception as error:

        return f"{type(error).__name__}: {error}"

    return None

def purge_fs_caches() -> dict[str, Any]:

    """Best-effort OS page-cache purge.  Never raises; honesty over success."""

    system = platform.system()

    if system == "Darwin":

        error = _run_purge_command([DARWIN_PURGE])

        return _purge_result(True, error is None, "darwin_purge", error)

    if system == "Linux":

        error = _drop_linux_caches()

        return _purge_result(True, error is None, "linux_drop_caches", error)

    return _purge_result(False, False, None, f"unsupported_platform: {system}")

def cache_fields(

    requested: str | None,

    purge_result: dict[str, Any] | None,

    *,

    warmed_by_scenario: bool = False,

) -> dict[str, Any]:

    """Honest ``cache_state``/``cache_state_reason`` row fields.

    ``warmed_by_scenario`` is the caller's explicit claim that it ran a warming

    pass; without that claim a "warm" request still reports "unknown".

    """

    if requested is None:

        return {"cache_state": "unknown", "cache_state_reason": "cache_state_unset"}

    if requested == "cold":

        if purge_result and purge_result.get("ok"):

            method = purge_result.get("method") or "unknown_method"

            return {"cache_state": "cold", "cache_state_reason": f"fs_caches_purged: {method}"}

        if purge_result and purge_result.get("attempted"):

            error = purge_result.get("error") or "unknown_error"

            return {"cache_state": "unknown", "cache_state_reason": f"purge_failed: {error}"}

        return {"cache_state": "unknown", "cache_state_reason": "purge_not_attempted"}

    if requested == "warm":

        if warmed_by_scenario:

            return {"cache_state": "warm", "cache_state_reason": "explicitly_warmed_by_scenario"}

        return {"cache_state": "unknown", "cache_state_reason": "warm_requested_without_warming_claim"}

    return {"cache_state": "unknown", "cache_state_reason": f"unrecognized_cache_state_request: {requested}"}