chemenu/tools/chemenu/telemetry/policy.py

"""Telemetry on/off and its two quantity caps, resolved once per root.

One place to answer three questions, because `writer.py`, `wikitool doctor`
and the MCP server's start-up guard must agree on the same answer for the
same checkout - a second copy of this logic is exactly how the MCP server
used to drift from the writer (it read `WIKI_TRACE` itself; fixed alongside
this module, see AGENTS.md invariant 8).

**Precedence for `enabled`:** `WIKI_TRACE` (either direction) beats
`.wikitool-telemetry.json`, which beats the installation-form default. The
form is read off `config.RELEASE_STAMP_FILENAME`: present means an exported/
distributed tree (default *off* - an operator never asked for telemetry),
absent means the dev checkout this package ships from (default *on* - the
traces are this stack's own measuring instrument, see
docs/explanation/telemetry-and-evaluation.md).

**Caching.** Only the filesystem-derived half - the release stamp and the
config file - is cached, keyed on the resolved root. `WIKI_TRACE` and the two
`WIKI_TRACE_MAX_SESSION_BYTES`/`WIKI_TRACE_KEEP_SESSIONS` overrides are read
fresh on every call: they are cheap (no I/O) and a test that flips one
mid-run must see the new value immediately, not a cached one. A test that
rewrites the release stamp or the config file at an already-resolved root
calls `reset_cache()` itself, the same discipline `conventions.reset_cache()`
already follows.

Stdlib only, like the rest of `chemenu.telemetry` - a hook handler imports
this package on every tool call and must not need the venv.
"""
from __future__ import annotations

import json
import os
from dataclasses import dataclass
from pathlib import Path

from chemenu import config, config_shapes

# The file's shape - and these two defaults with it - is the record in
# `chemenu.config_shapes`, which is stdlib-only for exactly this import.
DEFAULT_MAX_SESSION_BYTES = config_shapes.TELEMETRY.field("max_session_bytes").default
DEFAULT_KEEP_SESSIONS = config_shapes.TELEMETRY.field("keep_sessions").default

ENV_ENABLED = "WIKI_TRACE"
ENV_MAX_SESSION_BYTES = "WIKI_TRACE_MAX_SESSION_BYTES"
ENV_KEEP_SESSIONS = "WIKI_TRACE_KEEP_SESSIONS"


@dataclass(frozen=True)
class Policy:
    enabled: bool
    reason: str  # human-readable - what decided `enabled`, for `doctor`
    max_session_bytes: int
    keep_sessions: int


@dataclass(frozen=True)
class _RootDefaults:
    form_default: bool
    form_reason: str
    file_enabled: bool | None
    file_reason: str | None
    file_max_bytes: int | None
    file_keep_sessions: int | None


_cache: dict[Path, _RootDefaults] = {}


def reset_cache() -> None:
    _cache.clear()


def _positive_int(value: object) -> int | None:
    try:
        parsed = int(value)  # type: ignore[arg-type]
    except (TypeError, ValueError):
        return None
    return parsed if parsed > 0 else None


def _given(checked: "config_shapes.Checked", name: str) -> int | None:
    """A cap the file sets and the record accepts, else None - so the
    environment and the default decide, as for an absent file."""
    return checked.values[name] if name in checked.given else None


def _read_config(root: Path) -> dict:
    try:
        raw = (root / config.TELEMETRY_FILENAME).read_text(encoding="utf-8")
    except OSError:
        return {}
    try:
        data = json.loads(raw)
    except json.JSONDecodeError:
        return {}
    return data if isinstance(data, dict) else {}


def _root_defaults(root: Path) -> _RootDefaults:
    cached = _cache.get(root)
    if cached is not None:
        return cached

    distribution = (root / config.RELEASE_STAMP_FILENAME).is_file()
    form_default = not distribution
    form_reason = (
        f"no {config.RELEASE_STAMP_FILENAME} - dev checkout, default on"
        if form_default
        else f"{config.RELEASE_STAMP_FILENAME} present - distributed instance, default off"
    )

    # Silent on the hot path: a value the record does not accept is dropped,
    # never an error. `doctor` reports it (`config_shapes.deviations`).
    checked = config_shapes.check(config_shapes.TELEMETRY, _read_config(root))
    file_enabled = checked.values.get("enabled")
    file_reason = (
        f"{config.TELEMETRY_FILENAME} sets enabled={file_enabled}"
        if file_enabled is not None
        else None
    )

    result = _RootDefaults(
        form_default=form_default,
        form_reason=form_reason,
        file_enabled=file_enabled,
        file_reason=file_reason,
        file_max_bytes=_given(checked, "max_session_bytes"),
        file_keep_sessions=_given(checked, "keep_sessions"),
    )
    _cache[root] = result
    return result


def resolve(root: "Path | str | None" = None) -> Policy:
    """The effective policy for `root` (default: `config.ROOT`)."""
    resolved_root = Path(root).resolve() if root is not None else Path(config.ROOT).resolve()
    defaults = _root_defaults(resolved_root)

    env_enabled = os.environ.get(ENV_ENABLED)
    if env_enabled is not None:
        enabled = env_enabled != "0"
        reason = f"{ENV_ENABLED}={env_enabled!r} overrides"
    elif defaults.file_enabled is not None:
        enabled = defaults.file_enabled
        reason = defaults.file_reason  # type: ignore[assignment]
    else:
        enabled = defaults.form_default
        reason = defaults.form_reason

    max_bytes = (
        _positive_int(os.environ.get(ENV_MAX_SESSION_BYTES))
        or defaults.file_max_bytes
        or DEFAULT_MAX_SESSION_BYTES
    )
    keep_sessions = (
        _positive_int(os.environ.get(ENV_KEEP_SESSIONS))
        or defaults.file_keep_sessions
        or DEFAULT_KEEP_SESSIONS
    )

    return Policy(
        enabled=enabled,
        reason=reason,
        max_session_bytes=max_bytes,
        keep_sessions=keep_sessions,
    )