repo-manager/src/repo_manager/dual_run.py

"""Dual-run flags and checkout-mutation meter (RMGR-WP-0002/0003).

Precedence (highest first):
  1. Environment variables when set
  2. Config file (~/.repo-manager/dual-run.yaml or RM_DUAL_RUN_CONFIG)
  3. Defaults (all off)

Environment variables:
  RM_WRITEBACK, RM_RECONCILE, RM_WRITEBACK_PUSH, RM_PILOT_REPOS, RM_METER_PATH,
  RM_DUAL_RUN_CONFIG

Rollback: writeback/reconcile false in config and unset env → State Hub native.
"""

from __future__ import annotations

import json
import os
from datetime import datetime, timezone
from functools import lru_cache
from pathlib import Path
from typing import Any, Literal

import yaml

Source = Literal["state-hub", "repo-manager"]

_TRUE = frozenset({"1", "true", "yes", "on"})
_FALSE = frozenset({"0", "false", "no", "off"})


def config_path() -> Path:
    raw = os.environ.get("RM_DUAL_RUN_CONFIG", "~/.repo-manager/dual-run.yaml")
    return Path(raw).expanduser()


@lru_cache(maxsize=1)
def _load_file_config() -> dict[str, Any]:
    path = config_path()
    if not path.is_file():
        return {}
    try:
        data = yaml.safe_load(path.read_text(encoding="utf-8")) or {}
    except (OSError, yaml.YAMLError):
        return {}
    return data if isinstance(data, dict) else {}


def reload_config() -> None:
    """Clear cached config (tests / SIGHUP-style)."""
    _load_file_config.cache_clear()


def _env_bool(name: str) -> bool | None:
    raw = os.environ.get(name)
    if raw is None or raw.strip() == "":
        return None
    v = raw.strip().lower()
    if v in _TRUE:
        return True
    if v in _FALSE:
        return False
    return None


def _cfg_bool(key: str, default: bool = False) -> bool:
    env_map = {
        "writeback": "RM_WRITEBACK",
        "reconcile": "RM_RECONCILE",
        "writeback_push": "RM_WRITEBACK_PUSH",
    }
    env_name = env_map.get(key)
    if env_name:
        ev = _env_bool(env_name)
        if ev is not None:
            return ev
    cfg = _load_file_config()
    if key in cfg:
        val = cfg[key]
        if isinstance(val, bool):
            return val
        if isinstance(val, str):
            return val.strip().lower() in _TRUE
    return default


def writeback_enabled() -> bool:
    return _cfg_bool("writeback", False)


def reconcile_enabled() -> bool:
    return _cfg_bool("reconcile", False)


def writeback_push_enabled() -> bool:
    return _cfg_bool("writeback_push", False)


def pilot_slugs() -> set[str] | None:
    """None = all repos; empty set = none."""
    raw = os.environ.get("RM_PILOT_REPOS")
    if raw is not None:
        raw = raw.strip()
        if not raw:
            return set()
        return {s.strip() for s in raw.split(",") if s.strip()}
    cfg = _load_file_config()
    if "pilot_repos" not in cfg:
        return None
    val = cfg["pilot_repos"]
    if val is None:
        return None
    if isinstance(val, str):
        if not val.strip():
            return set()
        return {s.strip() for s in val.split(",") if s.strip()}
    if isinstance(val, list):
        return {str(s).strip() for s in val if str(s).strip()}
    return None


def slug_allowed(slug: str | None) -> bool:
    if not slug:
        return False
    pilots = pilot_slugs()
    if pilots is None:
        return True
    return slug in pilots


def writeback_for_repo(slug: str | None) -> bool:
    return writeback_enabled() and slug_allowed(slug)


def reconcile_for_repo(slug: str | None) -> bool:
    return reconcile_enabled() and slug_allowed(slug)


def meter_path() -> Path:
    raw = os.environ.get("RM_METER_PATH")
    if raw:
        return Path(raw).expanduser()
    cfg = _load_file_config()
    if cfg.get("meter_path"):
        return Path(str(cfg["meter_path"])).expanduser()
    return Path("~/.repo-manager/mutation-meter.jsonl").expanduser()


def record_mutation(
    *,
    source: Source,
    kind: str,
    repo_slug: str | None,
    detail: dict[str, Any] | None = None,
) -> None:
    try:
        path = meter_path()
        path.parent.mkdir(parents=True, exist_ok=True)
        row = {
            "ts": datetime.now(timezone.utc).isoformat(),
            "source": source,
            "kind": kind,
            "repo_slug": repo_slug,
            "detail": detail or {},
        }
        with path.open("a", encoding="utf-8") as fh:
            fh.write(json.dumps(row, default=str) + "\n")
    except OSError:
        pass


def meter_summary(path: Path | None = None) -> dict[str, int]:
    path = path or meter_path()
    counts: dict[str, int] = {"state-hub": 0, "repo-manager": 0, "total": 0}
    if not path.is_file():
        return counts
    for line in path.read_text(encoding="utf-8").splitlines():
        line = line.strip()
        if not line:
            continue
        try:
            row = json.loads(line)
        except json.JSONDecodeError:
            continue
        src = row.get("source")
        if src in counts:
            counts[src] += 1
        counts["total"] += 1
    return counts


def flags_status() -> dict[str, Any]:
    return {
        "config_path": str(config_path()),
        "config_exists": config_path().is_file(),
        "writeback": writeback_enabled(),
        "reconcile": reconcile_enabled(),
        "writeback_push": writeback_push_enabled(),
        "pilot_repos": sorted(pilot_slugs()) if pilot_slugs() is not None else None,
        "meter_path": str(meter_path()),
        "meter": meter_summary(),
        # legacy env-style keys for operators
        "RM_WRITEBACK": writeback_enabled(),
        "RM_RECONCILE": reconcile_enabled(),
        "RM_WRITEBACK_PUSH": writeback_push_enabled(),
        "RM_PILOT_REPOS": sorted(pilot_slugs()) if pilot_slugs() is not None else None,
        "RM_METER_PATH": str(meter_path()),
    }